|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
|
#ifndef EVENTFILTER_UTILTIES_PLUGINS_EVFDAQDIRECTOR
#define EVENTFILTER_UTILTIES_PLUGINS_EVFDAQDIRECTOR
#include "FWCore/ParameterSet/interface/ParameterSet.h"
#include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
#include "DataFormats/Provenance/interface/RunID.h"
#include "EventFilter/Utilities/interface/FFFNamingSchema.h"
#include "EventFilter/Utilities/interface/DirManager.h"
//std headers
#include <filesystem>
#include <iomanip>
#include <list>
#include <map>
#include <mutex>
#include <sstream>
#include <string>
#include <vector>
#include <functional>
//system headers
#include <sys/stat.h>
#include <sys/file.h>
#include <fcntl.h>
#include <cerrno>
#include <cstring>
#include <cstdio>
#include <boost/asio.hpp>
#include <oneapi/tbb/concurrent_hash_map.h>
typedef std::function<int(std::string const&, int&, int64_t&, uint32_t, bool&)> RawFileEvtCounter;
class SystemBounds;
class GlobalContext;
class StreamID;
namespace edm {
class ProcessContext;
} // namespace edm
namespace jsoncollector {
class DataPointDefinition;
namespace Json {
class Value;
}
} // namespace jsoncollector
namespace edm {
class ConfigurationDescriptions;
}
namespace evf {
enum MergeType { MergeTypeNULL = 0, MergeTypeDAT = 1, MergeTypePB = 2, MergeTypeJSNDATA = 3 };
class FastMonitoringService;
class EvFDaqDirector {
public:
enum FileStatus { noFile, sameFile, newFile, newLumi, runEnded, runAbort };
explicit EvFDaqDirector(const edm::ParameterSet& pset, edm::ActivityRegistry& reg);
~EvFDaqDirector();
void initRun();
static void fillDescriptions(edm::ConfigurationDescriptions& descriptions);
void preallocate(edm::service::SystemBounds const& bounds);
void preBeginRun(edm::GlobalContext const& globalContext);
void postEndRun(edm::GlobalContext const& globalContext);
void preGlobalEndLumi(edm::GlobalContext const& globalContext);
void updateRunParams();
void overrideRunNumber(unsigned int run) {
run_ = run;
updateRunParams();
}
std::string const& runString() const { return run_string_; }
std::string& baseRunDir() { return run_dir_; }
std::string& buBaseRunDir() { return bu_run_dir_; }
std::string& buBaseRunOpenDir() { return bu_run_open_dir_; }
bool useFileBroker() const { return useFileBroker_; }
std::string findCurrentRunDir() { return dirManager_.findRunDir(run_); }
std::string getInputJsonFilePath(const unsigned int ls, const unsigned int index) const;
std::string getRawFilePath(const unsigned int ls, const unsigned int index) const;
std::string getOpenRawFilePath(const unsigned int ls, const unsigned int index) const;
std::string getOpenInputJsonFilePath(const unsigned int ls, const unsigned int index) const;
std::string getDatFilePath(const unsigned int ls, std::string const& stream) const;
std::string getOpenDatFilePath(const unsigned int ls, std::string const& stream) const;
std::string getOpenOutputJsonFilePath(const unsigned int ls, std::string const& stream) const;
std::string getOutputJsonFilePath(const unsigned int ls, std::string const& stream) const;
std::string getMergedDatFilePath(const unsigned int ls, std::string const& stream) const;
std::string getMergedDatChecksumFilePath(const unsigned int ls, std::string const& stream) const;
std::string getOpenInitFilePath(std::string const& stream) const;
std::string getInitFilePath(std::string const& stream) const;
std::string getInitTempFilePath(std::string const& stream) const;
std::string getOpenProtocolBufferHistogramFilePath(const unsigned int ls, std::string const& stream) const;
std::string getProtocolBufferHistogramFilePath(const unsigned int ls, std::string const& stream) const;
std::string getMergedProtocolBufferHistogramFilePath(const unsigned int ls, std::string const& stream) const;
std::string getOpenRootHistogramFilePath(const unsigned int ls, std::string const& stream) const;
std::string getRootHistogramFilePath(const unsigned int ls, std::string const& stream) const;
std::string getMergedRootHistogramFilePath(const unsigned int ls, std::string const& stream) const;
std::string getEoLSFilePathOnBU(const unsigned int ls) const;
std::string getEoLSFilePathOnFU(const unsigned int ls) const;
std::string getBoLSFilePathOnFU(const unsigned int ls) const;
std::string getEoRFilePath() const;
std::string getEoRFileName() const;
std::string getEoRFilePathOnFU() const;
std::string getFFFParamsFilePathOnBU() const;
std::string getRunOpenDirPath() const { return run_dir_ + "/open"; }
bool outputAdler32Recheck() const { return outputAdler32Recheck_; }
void removeFile(std::string);
FileStatus updateFuLock(unsigned int& ls,
std::string& nextFile,
uint32_t& fsize,
uint16_t& rawHeaderSize,
uint64_t& lockWaitTime,
bool& setExceptionState);
void tryInitializeFuLockFile();
unsigned int getRunNumber() const { return run_; }
void lockInitLock();
void unlockInitLock();
void setFMS(evf::FastMonitoringService* fms) { fms_ = fms; }
bool isSingleStreamThread() { return nStreams_ == 1 && nThreads_ == 1; }
unsigned int numConcurrentLumis() const { return nConcurrentLumis_; }
void lockFULocal();
void unlockFULocal();
void lockFULocal2();
void unlockFULocal2();
void createBoLSFile(const uint32_t lumiSection, bool checkIfExists) const;
static int parseFRDFileHeader(std::string const& rawSourcePath,
int& rawFd,
uint16_t& rawHeaderSize,
uint16_t& rawDataType,
uint32_t& lsFromHeader,
int32_t& eventsFromHeader,
int64_t& fileSizeFromHeader,
bool requireHeader,
bool retry,
bool closeFile);
int grabNextJsonFromRaw(std::string const& rawSourcePath,
int& rawFd,
uint16_t& rawHeaderSize,
int64_t& fileSizeFromHeader,
bool& fileFound,
uint32_t serverLS,
bool closeFile,
bool requireHeader = true);
int grabNextJsonFile(std::string const& jsonSourcePath,
std::string const& rawSourcePath,
int64_t& fileSizeFromJson,
bool& fileFound);
int grabNextJsonFileAndUnlock(std::filesystem::path const& jsonSourcePath);
FileStatus getNextFromFileBroker(const unsigned int currentLumiSection,
unsigned int& ls,
std::string& nextFile,
int& rawFd,
uint16_t& rawHeaderSize,
int32_t& serverEventsInNewFile_,
int64_t& fileSize,
uint64_t& thisLockWaitTimeUs,
bool requireHeader = true,
bool fsDiscovery = false,
RawFileEvtCounter eventCounter = nullptr);
void createRunOpendirMaybe();
void createProcessingNotificationMaybe() const;
int readLastLSEntry(std::string const& file);
unsigned int getLumisectionToStart() const;
unsigned int getStartLumisectionFromEnv() const { return startFromLS_; }
std::string getStreamDestinations(std::string const&) const { return std::string(""); }
std::string getStreamMergeType(std::string const&, MergeType defaultType) const {
return MergeTypeNames_[defaultType];
}
static struct flock make_flock(short type, short whence, off_t start, off_t len, pid_t pid);
bool inputThrottled();
bool lumisectionDiscarded(unsigned int ls);
std::vector<std::string> const& getBUBaseDirs() const { return bu_base_dirs_all_; }
std::vector<int> const& getBUBaseDirsNSources() const { return bu_base_dirs_n_sources_; }
std::vector<int> const& getBUBaseDirsSourceIDs() const { return bu_base_dirs_source_ids_; }
std::string const& getSourceIdentifier() const { return source_identifier_; }
void setFileListMode() { fileListMode_ = true; }
bool fileListMode() const { return fileListMode_; }
unsigned int lsWithFilesOpen(unsigned int ls) const;
private:
void createLumiSectionFiles(const uint32_t lumiSection,
const uint32_t currentLumiSection,
bool doCreateBoLS,
bool doCreateEoLS);
bool rawFileHasHeader(std::string const& rawSourcePath, uint16_t& rawHeaderSize);
EvFDaqDirector::FileStatus contactFileBroker(unsigned int& serverHttpStatus,
bool& serverState,
uint32_t& serverLS,
uint32_t& closedServerLS,
std::string& nextFileJson,
std::string& nextFileRaw,
bool& rawHeader,
int maxLS);
EvFDaqDirector::FileStatus discoverFile(unsigned int& serverHttpStatus,
bool& serverState,
uint32_t& serverLS,
uint32_t& closedServerLS,
std::string& nextFileJson,
std::string& nextFileRaw,
bool& rawHeader,
int maxLS);
bool bumpFile(unsigned int& ls,
unsigned int& index,
std::string& nextFile,
uint32_t& fsize,
uint16_t& rawHeaderSize,
int maxLS,
bool& setExceptionState);
void openFULockfileStream(bool create);
static bool checkFileRead(char* buf, int& infile, std::size_t buf_sz, std::string const& path);
bool hasFRDFileHeader(std::string const& rawPath, int& rawFd, bool& hasErr, bool closeFile) const;
std::string inputFileNameStem(const unsigned int ls, const unsigned int index) const;
std::string outputFileNameStem(const unsigned int ls, std::string const& stream) const;
std::string mergedFileNameStem(const unsigned int ls, std::string const& stream) const;
std::string initFileName(std::string const& stream) const;
std::string eolsFileName(const unsigned int ls) const;
std::string eorFileName() const;
int getNFilesFromEoLS(std::string BUEoLSFile);
std::string base_dir_;
std::string bu_base_dir_;
std::vector<std::string> bu_base_dirs_all_;
std::vector<int> bu_base_dirs_n_sources_;
std::vector<int> bu_base_dirs_source_ids_;
std::string source_identifier_;
std::string sourceid_first_;
unsigned int run_;
bool useFileBroker_;
bool fileBrokerHostFromCfg_;
std::string fileBrokerHost_;
std::string fileBrokerPort_;
bool fileBrokerKeepAlive_;
bool fileBrokerUseLocalLock_;
unsigned int fuLockPollInterval_;
bool outputAdler32Recheck_;
bool directorBU_;
std::string hltSourceDirectory_;
unsigned int startFromLS_ = 1;
oneapi::tbb::concurrent_hash_map<unsigned int, unsigned int> lsWithFilesMap_;
std::string hostname_;
std::string run_string_;
std::string run_nstring_;
std::string pid_;
std::string run_dir_;
std::string bu_run_dir_;
std::string bu_run_open_dir_;
std::string fulockfile_;
int bu_readlock_fd_;
int bu_writelock_fd_;
int fu_readwritelock_fd_;
int fulocal_rwlock_fd_;
int fulocal_rwlock_fd2_;
FILE* bu_w_lock_stream;
FILE* bu_r_lock_stream;
FILE* fu_rw_lock_stream;
FILE* bu_w_monitor_stream;
FILE* bu_t_monitor_stream;
DirManager dirManager_;
unsigned long previousFileSize_;
struct flock bu_w_flk;
struct flock bu_r_flk;
struct flock bu_w_fulk;
struct flock bu_r_fulk;
struct flock fu_rw_flk;
struct flock fu_rw_fulk;
evf::FastMonitoringService* fms_ = nullptr;
pthread_mutex_t init_lock_ = PTHREAD_MUTEX_INITIALIZER;
unsigned int nStreams_ = 0;
unsigned int nThreads_ = 0;
unsigned int nConcurrentLumis_ = 0;
bool readEolsDefinition_ = true;
unsigned int eolsNFilesIndex_ = 1;
std::string stopFilePath_;
std::string stopFilePathPid_;
unsigned int stop_ls_override_ = 0;
//values initialized in .cc file
static const std::vector<std::string> MergeTypeNames_;
//json parser
jsoncollector::DataPointDefinition* dpd_;
boost::asio::io_context io_service_;
std::unique_ptr<boost::asio::ip::tcp::resolver> resolver_;
std::unique_ptr<boost::asio::ip::tcp::resolver::results_type> endpoint_iterator_;
std::unique_ptr<boost::asio::ip::tcp::socket> socket_;
std::string input_throttled_file_;
std::string discard_ls_filestem_;
bool fileListMode_ = false;
std::pair<unsigned, int> lastFileIdx_ = std::make_pair<unsigned, int>(0, -1);
};
} // namespace evf
#endif
|