5 #include <framework/logging/Logger.h>
6 #include <framework/pcore/EvtMessage.h>
8 #include <daq/storage/SharedEventBuffer.h>
10 #include <daq/slc/database/DBHandlerException.h>
12 #include <daq/slc/psql/PostgreSQLInterface.h>
14 #include <daq/slc/readout/RunInfoBuffer.h>
16 #include <daq/slc/base/ConfigFile.h>
17 #include <daq/slc/base/Date.h>
18 #include <daq/slc/base/StringUtil.h>
25 #include <sys/statvfs.h>
30 const unsigned long long GB = 1000 * 1024 * 1024;
31 const unsigned long long MAX_FILE_SIZE = 2 * GB;
32 const char* g_table =
"datafiles";
33 unsigned int g_streamersize = 0;
34 char* g_streamerinfo =
new char[1000000];
43 const char* host,
const char* dbtmp)
44 : m_db(db), m_runtype(runtype), m_host(host), m_dbtmp(dbtmp)
56 int getDiskId() {
return m_diskid; }
57 int getFileId() {
return m_fileid; }
60 int open(
const std::string& dir,
int ndisks,
int expno,
int runno,
int fileid)
71 bool available =
false;
72 for (
int i = 0; i < ndisks; i++) {
73 struct statvfs statfs;
75 sprintf(filename,
"%s%02d", dir.c_str(), m_diskid);
76 statvfs(filename, &statfs);
77 float usage = 1 - ((float)statfs.f_bfree / statfs.f_blocks);
79 sprintf(filename,
"%s%02d/storage/full_flag", dir.c_str(), m_diskid);
80 std::ifstream fin(filename);
85 std::cout <<
"[DEBUG] disk : " << m_diskid <<
" is available" << std::endl;
89 std::cout <<
"[DEBUG] disk : " << m_diskid <<
" is still full" << std::endl;
91 sprintf(filename,
"%s%02d/storage/full_flag", dir.c_str(), m_diskid);
92 std::ofstream fout(filename);
95 B2WARNING(
"disk-" << m_diskid <<
" is full " << usage);
98 if (m_diskid > ndisks) m_diskid = 1;
101 B2FATAL(
"No disk available for writing");
104 std::string filedir = dir + StringUtil::form(
"%02d/storage/%4.4d/%5.5d/",
105 m_diskid, m_expno, m_runno);
106 system((
"mkdir -p " + filedir).c_str());
107 m_filename = StringUtil::form(
"%s.%4.4d.%5.5d.%s.f%5.5d.sroot",
108 m_runtype.c_str(), m_expno, m_runno, m_host.c_str(), m_fileid);
109 m_path = filedir + m_filename;
110 m_file =
::open(
m_path.c_str(), O_WRONLY | O_CREAT | O_EXCL, 0664);
111 if (g_diskid > 0 && g_diskid != m_diskid) {
112 B2FATAL(
"disk-" << m_diskid <<
" is already full! Terminating process..");
117 B2FATAL(
"Failed to open file : " <<
m_path);
122 m_db->execute(
"insert into %s (name, path, host, label, expno, runno, fileno, nevents, chksum, size) "
123 "values ('%s', '%s', '%s', '%s', %d, %d, %d, 0, 0, 0);",
124 g_table, m_filename.c_str(),
m_path.c_str(), m_host.c_str(),
125 m_runtype.c_str(), m_expno, m_runno, m_fileid);
129 write(g_streamerinfo, g_streamersize,
true);
130 B2INFO(
"New file " <<
m_path <<
" is opened");
138 std::cout <<
"[DEBUG] File closed" << std::endl;
142 stat(
m_path.c_str(), &st);
143 std::string d =
Date(st.st_mtime).toString();
146 m_db->execute(
"update %s set time_close = '%s', chksum = %lu, nevents = %lu, "
147 "size = %lu where name = '%s' and host = '%s';",
148 g_table, d.c_str(), m_chksum, m_nevents, m_filesize,
149 m_filename.c_str(), m_host.c_str());
157 int write(
char* evtbuf,
int nbyte,
bool isstreamer =
false)
159 int ret = ::write(m_file, evtbuf, nbyte);
164 m_chksum = adler32(m_chksum, (
unsigned char*)evtbuf, nbyte);
175 std::string m_runtype;
179 std::string m_filename;
184 std::string m_configname;
187 unsigned long long m_filesize;
188 unsigned long long m_chksum;
189 unsigned long long m_nevents;
194 void signalHandler(
int)
196 if (g_file) g_file->close();
200 int main(
int argc,
char** argv)
203 printf(
"%s : <ibufname> <ibufsize> <hostname> <runtype> <path> <ndisk> "
204 "<filepath_dbtmp> [<obufname> <obufsize> nodename, nodeid]\n", argv[0]);
208 const unsigned interval = 1;
209 const char* ibufname = argv[1];
210 const int ibufsize = atoi(argv[2]);
211 const char* hostname = argv[3];
212 const char* runtype = argv[4];
213 const bool not_record = std::string(runtype) ==
"null";
214 const char* path = argv[5];
215 const int ndisks = atoi(argv[6]);
216 const char* file_dbtmp = argv[7];
217 const char* obufname = (argc > 7) ? argv[8] :
"";
218 const int obufsize = (argc > 8) ? atoi(argv[9]) : -1;
219 const char* nodename = (argc > 9) ? argv[10] :
"";
220 const int nodeid = (argc > 10) ? atoi(argv[11]) : -1;
221 const unsigned int ninput = (argc > 11) ? atoi(argv[12]) : 1;
224 const bool use_info = nodeid >= 0;
226 info.open(nodename, nodeid);
229 for (
unsigned int ib = 0; ib < ninput; ib++) {
230 ibuf[ib].open(StringUtil::form(
"%s_%d", ibufname, ib), ibufsize * 1000000);
232 signal(SIGINT, signalHandler);
233 signal(SIGKILL, signalHandler);
236 config.get(
"database.dbname"),
237 config.get(
"database.user"),
238 config.get(
"database.password"),
239 config.getInt(
"database.port"));
241 if (obufsize > 0) obuf.open(obufname, obufsize * 1000000);
242 if (use_info) info.reportReady();
243 B2DEBUG(1,
"started recording.");
244 unsigned long long nbyte_out = 0;
245 unsigned int count_out = 0;
246 unsigned int expno = 0;
247 unsigned int runno = 0;
248 unsigned int subno = 0;
249 int* evtbuf =
new int[10000000];
250 g_file =
new FileHandler(db, runtype, hostname, file_dbtmp);
254 unsigned int fileid = 0;
263 if (use_info) info.reportRunning();
265 while ((g_runno = info.getRunNumber()) <= 0) {
268 g_expno = info.getExpNumber();
269 std::cout <<
"[DEBUG] expno = " << g_expno <<
", runno =" << g_runno << std::endl;
271 for (
unsigned int ib = 0; ib < ninput; ib++) {
273 ibuf[ib].read((
int*)&hd,
true,
true);
274 ibuf[ib].read(evtbuf,
true,
true);
276 int nbyte = evtbuf[0];
277 int nword = (nbyte - 1) / 4 + 1;
279 if (not_record)
continue;
280 if (hd.type == MSG_STREAMERINFO) {
281 memcpy(g_streamerinfo, evtbuf, nbyte);
282 g_streamersize = nbyte;
284 if (expno > hd.expno || runno > hd.runno) {
285 B2WARNING(
"Old run was detected => discard event exp = " << hd.expno <<
" (" << expno <<
"), runno" << hd.runno <<
"(" << runno <<
289 if (!newrun || expno < hd.expno || runno < hd.runno) {
296 info.setExpNumber(expno);
297 info.setRunNumber(runno);
298 info.setSubNumber(subno);
299 info.setInputCount(0);
300 info.setInputNBytes(0);
301 info.setOutputCount(0);
302 info.setOutputNBytes(0);
308 oheader->expno = expno;
309 oheader->runno = runno;
314 file.open(path, ndisks, expno, runno, fileid);
320 info.addInputCount(1);
321 info.addInputNBytes(nbyte);
323 if (hd.type == MSG_STREAMERINFO) {
335 file.write((
char*)evtbuf, nbyte);
337 if (!isnew && obufsize > 0 && count_out % interval == 0 && obuf.isWritable(nword)) {
338 obuf.write(evtbuf, nword,
true);
342 info.addOutputCount(1);
343 info.addOutputNBytes(nbyte);
344 info.get()->reserved[0] = file.getFileId();
345 info.get()->reserved[1] = file.getDiskId();
346 info.get()->reserved_f[0] = (float)info.getOutputNBytes() / 1024. / 1024.;
350 B2WARNING(
"no run was initialzed for recording : " << hd.expno <<
"." << hd.runno);