Gaudi Framework, version v23r9
Home   Generated: Thu Jul 18 2013
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Groups Pages
IODataManager.cpp
Go to the documentation of this file.
1 // Framework include files
2 #include "GaudiKernel/Debugger.h"
3 #include "GaudiKernel/MsgStream.h"
4 #include "GaudiKernel/strcasecmp.h"
5 #include "GaudiKernel/DeclareFactoryEntries.h"
6 #include "GaudiUtils/IFileCatalog.h"
7 #include "IODataManager.h"
8 #include "GaudiKernel/SmartIF.h"
9 #include "GaudiKernel/Incident.h"
10 #include "GaudiKernel/IIncidentSvc.h"
11 
12 #include <set>
13 
14 DECLARE_NAMESPACE_SERVICE_FACTORY(Gaudi,IODataManager)
15 
16 using namespace Gaudi;
17 
18 enum { S_OK = StatusCode::SUCCESS, S_ERROR=StatusCode::FAILURE };
19 
20 static std::set<std::string> s_badFiles;
21 
22 IODataManager::IODataManager(CSTR nam, ISvcLocator* svcloc)
23  : base_class(nam, svcloc), m_ageLimit(2)
24 {
25  declareProperty("CatalogType", m_catalogSvcName="Gaudi::MultiFileCatalog/FileCatalog");
26  declareProperty("UseGFAL", m_useGFAL = true);
27  declareProperty("QuarantineFiles", m_quarantine = true);
28  declareProperty("AgeLimit", m_ageLimit = 2);
29  declareProperty("DisablePFNWarning", m_disablePFNWarning = false,
30  "if set to True, we will not report when a file "
31  "is opened by it's physical name");
32 }
33 
35 StatusCode IODataManager::initialize() {
36  // Initialize base class
37  StatusCode status = Service::initialize();
38  MsgStream log(msgSvc(), name());
39  if ( !status.isSuccess() ) {
40  log << MSG::ERROR << "Error initializing base class Service!" << endmsg;
41  return status;
42  }
43  // Retrieve conversion service handling event iteration
44  m_catalog = serviceLocator()->service(m_catalogSvcName);
45  if( !m_catalog.isValid() ) {
46  log << MSG::ERROR
47  << "Unable to localize interface IFileCatalog from service:"
48  << m_catalogSvcName << endmsg;
49  return StatusCode::FAILURE;
50  }
51  m_incSvc = serviceLocator()->service("IncidentSvc");
52  if( !m_incSvc.isValid() ) {
53  log << MSG::ERROR << "Error initializing IncidentSvc Service!" << endmsg;
54  return status;
55  }
56 
57  return status;
58 }
59 
61 StatusCode IODataManager::finalize() {
62  m_catalog = 0; // release
63  return Service::finalize();
64 }
65 
66 // Small routine to issue exceptions
67 StatusCode IODataManager::error(CSTR msg, bool rethrow) {
68  MsgStream log(msgSvc(),name());
69  log << MSG::ERROR << "Error: " << msg << endmsg;
70  if ( rethrow ) {
71  System::breakExecution();
72  }
73  return S_ERROR;
74 }
76 IODataManager::Connections IODataManager::connections(const IInterface* owner) const {
77  Connections conns;
78  for(ConnectionMap::const_iterator i=m_connectionMap.begin(); i!=m_connectionMap.end();++i) {
79  IDataConnection* c = (*i).second->connection;
80  if ( 0 == owner || c->owner() == owner )
81  conns.push_back(c);
82  }
83  return conns;
84 }
85 
87 StatusCode IODataManager::connectRead(bool keep_open, Connection* con) {
88  if ( !establishConnection(con) ) {
89  return connectDataIO(UNKNOWN,Connection::READ,con->name(),"UNKNOWN",keep_open,con);
90  }
91  std::string dsn = con ? con->name() : std::string("Unknown");
92  return error("Failed to connect to data:"+dsn,false);
93 }
94 
96 StatusCode IODataManager::connectWrite(Connection* con,IoType mode,CSTR doctype) {
97  if ( !establishConnection(con) ) {
98  return connectDataIO(UNKNOWN,mode,con->name(),doctype,true,con);
99  }
100  std::string dsn = con ? con->name() : std::string("Unknown");
101  return error("Failed to connect to data:"+dsn,false);
102 }
103 
105 StatusCode IODataManager::read(Connection* con, void* const data, size_t len) {
106  return establishConnection(con).isSuccess() ? con->read(data,len) : S_ERROR;
107 }
108 
110 StatusCode IODataManager::write(Connection* con, const void* data, int len) {
111  return establishConnection(con).isSuccess() ? con->write(data,len) : S_ERROR;
112 }
113 
115 long long int IODataManager::seek(Connection* con, long long int where, int origin) {
116  return establishConnection(con).isSuccess() ? con->seek(where,origin) : -1;
117 }
118 
119 StatusCode IODataManager::disconnect(Connection* con) {
120  if ( con ) {
121  std::string dataset = con->name();
122  std::string dsn = dataset;
123  StatusCode sc = con->disconnect();
124  if ( ::strncasecmp(dsn.c_str(),"FID:",4)==0 )
125  dsn = dataset.substr(4);
126  else if ( ::strncasecmp(dsn.c_str(),"LFN:",4)==0 )
127  dsn = dataset.substr(4);
128  else if ( ::strncasecmp(dsn.c_str(),"PFN:",4)==0 )
129  dsn = dataset.substr(4);
130 
131  FidMap::iterator j = m_fidMap.find(dataset);
132  if ( j != m_fidMap.end() ) {
133  std::string fid = (*j).second;
134  std::string gfal_name = "gfal:guid:" + fid;
135  ConnectionMap::iterator i=m_connectionMap.find(fid);
136  m_fidMap.erase(j);
137  if ( (j=m_fidMap.find(fid)) != m_fidMap.end() )
138  m_fidMap.erase(j);
139  if ( (j=m_fidMap.find(gfal_name)) != m_fidMap.end() )
140  m_fidMap.erase(j);
141  if ( i != m_connectionMap.end() ) {
142  if ( (*i).second ) {
143  IDataConnection* c = (*i).second->connection;
144  std::string pfn = c->pfn();
145  if ( (j=m_fidMap.find(pfn)) != m_fidMap.end() )
146  m_fidMap.erase(j);
147  if ( c->isConnected() ) {
148  MsgStream log(msgSvc(),name());
149  c->disconnect();
150  log << MSG::INFO << "Disconnect from dataset " << dsn
151  << " [" << fid << "]" << endmsg;
152  }
153  delete (*i).second;
154  m_connectionMap.erase(i);
155  }
156  }
157  }
158  return sc;
159  }
160  return S_ERROR;
161 }
162 
163 StatusCode IODataManager::reconnect(Entry* e) {
164  StatusCode sc = S_ERROR;
165  if ( e && e->connection ) {
166  switch(e->ioType) {
167  case Connection::READ:
168  sc = e->connection->connectRead();
169  break;
170  case Connection::UPDATE:
171  case Connection::CREATE:
172  case Connection::RECREATE:
173  sc = e->connection->connectWrite(e->ioType);
174  break;
175  default:
176  return S_ERROR;
177  }
178  if ( sc.isSuccess() && e->ioType == Connection::READ ) {
179  std::vector<Entry*> to_retire;
180  e->connection->resetAge();
181  for(ConnectionMap::iterator i=m_connectionMap.begin(); i!=m_connectionMap.end();++i) {
182  IDataConnection* c = (*i).second->connection;
183  if ( e->connection != c && c->isConnected() && !(*i).second->keepOpen ) {
184  c->ageFile();
185  if ( c->age() > m_ageLimit ) {
186  to_retire.push_back((*i).second);
187  }
188  }
189  }
190  if ( !to_retire.empty() ) {
191  MsgStream log(msgSvc(),name());
192  for(std::vector<Entry*>::iterator j=to_retire.begin(); j!=to_retire.end();++j) {
193  IDataConnection* c = (*j)->connection;
194  c->disconnect();
195  log << MSG::INFO << "Disconnect from dataset " << c->pfn()
196  << " [" << c->fid() << "]" << endmsg;
197  }
198  }
199  }
200  }
201  return sc;
202 }
203 
205 IIODataManager::Connection* IODataManager::connection(CSTR dataset) const {
206  FidMap::const_iterator j = m_fidMap.find(dataset);
207  if ( j != m_fidMap.end() ) {
208  ConnectionMap::const_iterator i=m_connectionMap.find((*j).second);
209  return (i != m_connectionMap.end()) ? (*i).second->connection : 0;
210  }
211  return 0;
212 }
213 
214 StatusCode IODataManager::establishConnection(Connection* con) {
215  if ( con ) {
216  if ( !con->isConnected() ) {
217  ConnectionMap::const_iterator i=m_connectionMap.find(con->name());
218  if ( i != m_connectionMap.end() ) {
219  Connection* c = (*i).second->connection;
220  if ( c != con ) {
221  m_incSvc->fireIncident(Incident(con->name(),IncidentType::FailInputFile));
222  return error("Severe logic bug: Twice identical connection object for DSN:"+con->name(),true);
223  }
224  if ( reconnect((*i).second).isSuccess() ) {
225  return S_OK;
226  }
227  }
228  return S_ERROR;
229  }
230  con->resetAge();
231  return S_OK;
232  }
233  return error("Severe logic bug: No connection object avalible.",true);
234 }
235 
236 StatusCode
237 IODataManager::connectDataIO(int typ, IoType rw, CSTR dataset, CSTR technology,bool keep_open,Connection* connection) {
238  MsgStream log(msgSvc(),name());
239  std::string dsn = dataset;
240  try {
241  StatusCode sc(StatusCode::SUCCESS,true);
242  if ( ::strncasecmp(dsn.c_str(),"FID:",4)==0 )
243  dsn = dataset.substr(4), typ = FID;
244  else if ( ::strncasecmp(dsn.c_str(),"LFN:",4)==0 )
245  dsn = dataset.substr(4), typ = LFN;
246  else if ( ::strncasecmp(dsn.c_str(),"PFN:",4)==0 )
247  dsn = dataset.substr(4), typ = PFN;
248  else if ( typ == UNKNOWN )
249  return connectDataIO(PFN, rw, dsn, technology, keep_open, connection);
250 
251  if(std::find(s_badFiles.begin(),s_badFiles.end(),dsn) != s_badFiles.end()) {
252  m_incSvc->fireIncident(Incident(dsn,IncidentType::FailInputFile));
253  return IDataConnection::BAD_DATA_CONNECTION;
254  }
255  if ( typ == FID ) {
256  ConnectionMap::iterator fi = m_connectionMap.find(dsn);
257  if ( fi == m_connectionMap.end() ) {
258  IFileCatalog::Files files;
259  m_catalog->getPFN(dsn,files);
260  if ( files.size() == 0 ) {
261  if ( !m_useGFAL ) {
262  if ( m_quarantine ) s_badFiles.insert(dsn);
263  m_incSvc->fireIncident(Incident(dsn,IncidentType::FailInputFile));
264  error("connectDataIO> failed to resolve FID:"+dsn,false).ignore();
265  return IDataConnection::BAD_DATA_CONNECTION;
266  }
267  else if ( dsn.length() == 36 && dsn[8]=='-' && dsn[13]=='-' ) {
268  std::string gfal_name = "gfal:guid:" + dsn;
269  m_fidMap[dsn] = m_fidMap[dataset] = m_fidMap[gfal_name] = dsn;
270  sc = connectDataIO(PFN, rw, gfal_name, technology, keep_open, connection);
271  if ( sc.isSuccess() ) return sc;
272  if ( m_quarantine ) s_badFiles.insert(dsn);
273  }
274  if ( m_quarantine ) s_badFiles.insert(dsn);
275  m_incSvc->fireIncident(Incident(dsn,IncidentType::FailInputFile));
276  error("connectDataIO> Failed to resolve FID:"+dsn,false).ignore();
277  return IDataConnection::BAD_DATA_CONNECTION;
278  }
279  std::string pfn = files[0].first;
280  m_fidMap[dsn] = m_fidMap[dataset] = m_fidMap[pfn] = dsn;
281  sc = connectDataIO(PFN, rw, pfn, technology, keep_open, connection);
282  if ( !sc.isSuccess() ) {
283  if ( m_quarantine ) s_badFiles.insert(pfn);
284  m_incSvc->fireIncident(Incident(pfn,IncidentType::FailInputFile));
285  return IDataConnection::BAD_DATA_CONNECTION;
286  }
287 
288  return sc;
289  }
290  return S_ERROR;
291  //Connection* c = (*fi).second->connection;
292  //sc = connectDataIO(PFN, rw, c->pfn(), technology, keep_open, connection);
293  //if ( !sc.isSuccess() && m_quarantine ) s_badFiles.insert(c->pfn());
294  //return sc;
295  }
296  std::string fid;
297  FidMap::iterator j = m_fidMap.find(dsn);
298  if ( j == m_fidMap.end() ) {
299  IFileCatalog::Files files;
300  switch(typ) {
301  case LFN:
302  fid = m_catalog->lookupLFN(dsn);
303  if ( fid.empty() ) {
304  m_incSvc->fireIncident(Incident(dsn,IncidentType::FailInputFile));
305  log << MSG::ERROR << "Failed to resolve LFN:" << dsn
306  << " Cannot access this dataset." << endmsg;
307  return IDataConnection::BAD_DATA_CONNECTION;
308  }
309  break;
310  case PFN:
311  fid = m_catalog->lookupPFN(dsn);
312  if ( !fid.empty() ) m_catalog->getPFN(fid, files);
313  if ( files.empty() ) {
314  if ( rw == Connection::CREATE || rw == Connection::RECREATE ) {
315  if ( fid.empty() ) fid = m_catalog->createFID();
316  m_catalog->registerPFN(fid,dsn,technology);
317  log << MSG::INFO << "Referring to dataset " << dsn
318  << " by its file ID:" << fid << endmsg;
319  }
320  else {
321  fid = dsn;
322  }
323  }
324  break;
325  }
326  }
327  else {
328  fid = (*j).second;
329  }
330  if ( typ == PFN ) {
331  // Open PFN
332  ConnectionMap::iterator fi = m_connectionMap.find(fid);
333  if ( fi == m_connectionMap.end() ) {
334  connection->setFID(fid);
335  connection->setPFN(dsn);
336  Entry* e = new Entry(technology, keep_open, rw, connection);
337  // Here we open the file!
338  if ( !reconnect(e).isSuccess() ) {
339  delete e;
340  if ( m_quarantine ) s_badFiles.insert(dsn);
341  m_incSvc->fireIncident(Incident(dsn,IncidentType::FailInputFile));
342  error("connectDataIO> Cannot connect to database: PFN="+dsn+" FID="+fid,false).ignore();
343  return IDataConnection::BAD_DATA_CONNECTION;
344  }
345  fid = connection->fid();
346  m_fidMap[dataset] = m_fidMap[dsn] = m_fidMap[fid] = fid;
347  if ( !(rw==Connection::CREATE || rw==Connection::RECREATE) ) {
348  if ( ! m_disablePFNWarning && strcasecmp(dsn.c_str(),fid.c_str()) == 0 ) {
349  log << MSG::ERROR << "Referring to existing dataset " << dsn
350  << " by its physical name." << endmsg;
351  log << "You may not be able to navigate back to the input file"
352  << " -- processing continues" << endmsg;
353  }
354  }
355  m_connectionMap.insert(std::make_pair(fid,e));
356  return S_OK;
357  }
358  // Here we open the file!
359  if ( !reconnect((*fi).second).isSuccess() ) {
360  if ( m_quarantine ) s_badFiles.insert(dsn);
361  m_incSvc->fireIncident(Incident(dsn,IncidentType::FailInputFile));
362  error("connectDataIO> Cannot connect to database: PFN="+dsn+" FID="+fid,false).ignore();
363  return IDataConnection::BAD_DATA_CONNECTION;
364  }
365  return S_OK;
366  }
367  sc = connectDataIO(FID, rw, fid, technology, keep_open, connection);
368  if ( !sc.isSuccess() && m_quarantine ) {
369  s_badFiles.insert(fid);
370  }
371  else if ( typ == LFN ) {
372  m_fidMap[dataset] = fid;
373  }
374  return sc;
375  }
376  catch (std::exception& e) {
377  error(std::string("connectDataIO> Caught exception:")+e.what(), false).ignore();
378  }
379  catch(...) {
380  error(std::string("connectDataIO> Caught unknown exception"), false).ignore();
381  }
382  m_incSvc->fireIncident(Incident(dsn,IncidentType::FailInputFile));
383  error("connectDataIO> The dataset "+dsn+" cannot be opened.",false).ignore();
384  s_badFiles.insert(dsn);
385  return IDataConnection::BAD_DATA_CONNECTION;
386 }
Generated at Thu Jul 18 2013 12:18:05 for Gaudi Framework, version v23r9 by Doxygen version 1.8.2 written by Dimitri van Heesch, © 1997-2004