6 #include <framework/logging/Logger.h>
7 #include <framework/pcore/EvtMessage.h>
9 #include <daq/storage/SharedEventBuffer.h>
11 #include <daq/slc/psql/PostgreSQLInterface.h>
13 #include <daq/slc/database/DBHandlerException.h>
15 #include <daq/slc/readout/RunInfoBuffer.h>
17 #include <daq/slc/base/ConfigFile.h>
18 #include <daq/slc/base/Date.h>
19 #include <daq/slc/base/StringUtil.h>
26 #include <sys/statvfs.h>
31 const unsigned long long GB = 1000 * 1024 * 1024;
32 const unsigned long long MAX_FILE_SIZE = 8 * GB;
33 const char* g_table =
"datafiles";
34 unsigned int g_streamersize = 0;
35 char* g_streamerinfo =
new char[1000000];
37 std::string g_file_diskid;
40 bool g_is_arich =
false;
46 const char* host,
const char* dbtmp)
47 : m_db(db), m_runtype(runtype), m_host(host), m_dbtmp(dbtmp)
59 int getDiskId() {
return m_diskid; }
60 int getFileId() {
return m_fileid; }
63 int open(
const std::string& dir,
int ndisks,
int expno,
int runno,
int fileid)
71 m_expno = (g_expno > 0) ? g_expno : expno;
72 m_runno = (g_runno > 0) ? g_runno : runno;
74 bool available =
false;
76 struct statvfs statfs;
83 std::ifstream fin(g_file_diskid.c_str());
87 for (
int i = 0; i < ndisks; i++) {
88 sprintf(filename,
"%s%02d", dir.c_str(), m_diskid);
89 statvfs(filename, &statfs);
90 float usage = 1 - ((float)statfs.f_bfree / statfs.f_blocks);
91 sprintf(filename,
"%s%02d/storage/full_flag", dir.c_str(), m_diskid);
92 std::ifstream fin(filename);
111 std::cout <<
"[DEBUG] disk : " << m_diskid <<
" is available" << std::endl;
114 std::cout <<
"[DEBUG] disk : " << m_diskid <<
" is with full_flag" << std::endl;
116 std::ofstream fout(filename);
119 B2WARNING(
"disk-" << m_diskid <<
" is full " << usage);
122 if (m_diskid > ndisks) m_diskid = 1;
126 B2FATAL(
"No disk available for writing " << __FILE__ <<
":" << __LINE__);
129 std::string filedir = dir + StringUtil::form(
"%02d/storage/%4.4d/%5.5d/", m_diskid, expno, runno);
130 system((
"mkdir -p " + filedir).c_str());
131 m_filename = StringUtil::form(
"%s.%4.4d.%5.5d.%s.f%5.5d.sroot",
132 m_runtype.c_str(), expno, runno, m_host.c_str(), m_fileid);
133 m_path = filedir + m_filename;
134 m_file =
::open(
m_path.c_str(), O_WRONLY | O_CREAT | O_EXCL, 0664);
136 std::ofstream fout(g_file_diskid.c_str());
139 B2FATAL(
"Failed to open file : " <<
m_path);
144 m_db->execute(
"insert into %s (name, path, host, label, expno, runno, fileno, nevents, chksum, size) "
145 "values ('%s', '%s', '%s', '%s', %d, %d, %d, 0, 0, 0);",
146 g_table, m_filename.c_str(),
m_path.c_str(), m_host.c_str(),
147 m_runtype.c_str(), m_expno, m_runno, m_fileid);
151 write(g_streamerinfo, g_streamersize,
true);
152 B2INFO(
"New file " <<
m_path <<
" is opened");
160 std::cout <<
"[DEBUG] File closed" << std::endl;
164 stat(
m_path.c_str(), &st);
165 std::string d =
Date(st.st_mtime).toString();
167 m_db->execute(
"update %s set time_close = '%s', chksum = %lu, nevents = %lu, "
168 "size = %lu where name = '%s' and host = '%s';",
169 g_table, d.c_str(), m_chksum, m_nevents, m_filesize,
170 m_filename.c_str(), m_host.c_str());
178 int write(
char* evtbuf,
int nbyte,
bool isstreamer =
false)
180 m_chksum = adler32(m_chksum, (
unsigned char*)evtbuf, nbyte);
181 int ret = ::write(m_file, evtbuf, nbyte);
196 std::string m_runtype;
200 std::string m_filename;
205 std::string m_configname;
208 unsigned long long m_filesize;
209 unsigned long long m_chksum;
210 unsigned long long m_nevents;
215 void signalHandler(
int)
217 if (g_file) g_file->close();
221 int main(
int argc,
char** argv)
224 printf(
"%s : <ibufname> <ibufsize> <hostname> <runtype> <path> <ndisk> "
225 "<filepath_dbtmp> [<obufname> <obufsize> nodename, nodeid]\n", argv[0]);
228 const unsigned interval = 1;
229 const char* ibufname = argv[1];
230 const int ibufsize = atoi(argv[2]);
231 const char* hostname = argv[3];
232 const char* runtype = argv[4];
233 const bool not_record = std::string(runtype) ==
"null";
234 const char* path = argv[5];
235 const int ndisks = atoi(argv[6]);
236 const char* file_dbtmp = argv[7];
237 const char* obufname = (argc > 7) ? argv[8] :
"";
238 const int obufsize = (argc > 8) ? atoi(argv[9]) : -1;
239 const char* nodename = (argc > 9) ? argv[10] :
"";
240 g_is_arich = StringUtil::find(runtype,
"arich");
241 const int nodeid = (argc > 10) ? atoi(argv[11]) : -1;
242 const unsigned int ninput = (argc > 11) ? atoi(argv[12]) : 1;
243 g_file_diskid = StringUtil::form(
"/tmp/%s_diskid", nodename);
246 const bool use_info = nodeid >= 0;
248 info.open(nodename, nodeid);
251 for (
unsigned int ib = 0; ib < ninput; ib++) {
252 ibuf[ib].open(StringUtil::form(
"%s_%d", ibufname, ib), ibufsize * 1000000);
254 signal(SIGINT, signalHandler);
255 signal(SIGKILL, signalHandler);
258 config.get(
"database.dbname"),
259 config.get(
"database.user"),
260 config.get(
"database.password"),
261 config.getInt(
"database.port"));
263 if (obufsize > 0) obuf.open(obufname, obufsize * 1000000);
264 if (use_info) info.reportReady();
265 B2DEBUG(1,
"started recording.");
266 unsigned long long nbyte_out = 0;
267 unsigned int count_out = 0;
268 unsigned int expno = 0;
269 unsigned int runno = 0;
270 unsigned int subno = 0;
271 int* evtbuf =
new int[10000000];
272 g_file =
new FileHandler(db, runtype, hostname, file_dbtmp);
276 unsigned int fileid = 0;
285 if (use_info) info.reportRunning();
288 while ((g_runno = info.getRunNumber()) <= 0) {
291 g_expno = info.getExpNumber();
292 std::cout <<
"[DEBUG] expno = " << g_expno <<
", runno =" << g_runno << std::endl;
295 for (
unsigned int ib = 0; ib < ninput; ib++) {
297 ibuf[ib].read((
int*)&hd,
true,
true);
298 ibuf[ib].read(evtbuf,
true,
true);
300 int nbyte = evtbuf[0];
301 int nword = (nbyte - 1) / 4 + 1;
303 if (hd.type == MSG_STREAMERINFO) {
304 memcpy(g_streamerinfo, evtbuf, nbyte);
305 g_streamersize = nbyte;
307 if (expno > hd.expno || runno > hd.runno) {
308 B2WARNING(
"The old run was detected => discard event exp = " << hd.expno <<
309 " (" << expno <<
"), runno" << hd.runno <<
"(" << runno <<
")");
312 if (!newrun || expno < hd.expno || runno < hd.runno) {
319 info.setExpNumber(expno);
320 info.setRunNumber(runno);
321 info.setSubNumber(subno);
322 info.setInputCount(0);
323 info.setInputNBytes(0);
324 info.setOutputCount(0);
325 info.setOutputNBytes(0);
331 oheader->expno = expno;
332 oheader->runno = runno;
338 std::string
filename = StringUtil::form(
"%s%02d", path, g_diskid);
339 struct statvfs statfs;
341 float usage = 1 - ((float)statfs.f_bfree / statfs.f_blocks);
342 if (usage > 0.9) g_diskid = 0;
343 file.open(path, ndisks, expno, runno, fileid);
351 info.addInputCount(1);
352 info.addInputNBytes(nbyte);
354 if (hd.type == MSG_STREAMERINFO) {
358 if (nbyte_out > MAX_FILE_SIZE) {
362 std::string
filename = StringUtil::form(
"%s%02d", path, g_diskid);
363 struct statvfs statfs;
365 float usage = 1 - ((float)statfs.f_bfree / statfs.f_blocks);
367 B2WARNING(
"disk-" << g_diskid <<
" is already full. Stopping Run#" << runno);
368 std::cout <<
"[STOP=RUNCONTROL]" << std::endl;
372 file.open(path, ndisks, expno, runno, fileid);
376 file.write((
char*)evtbuf, nbyte);
379 info.addOutputCount(1);
380 info.addOutputNBytes(nbyte);
381 info.get()->reserved[0] = file.getFileId();
382 info.get()->reserved[1] = file.getDiskId();
383 info.get()->reserved_f[0] = (float)info.getOutputNBytes() / 1024. / 1024.;
387 B2WARNING(
"no run was initialzed for recording : " << hd.expno <<
"." << hd.runno);
392 if (!isnew && obufsize > 0 && count_out % interval == 0 && obuf.isWritable(nword)) {
393 obuf.write(evtbuf, nword,
true);