#include "InputFileCatalog.h"

Inheritance diagram for art::InputFileCatalog:

Classes
struct	Config

Public Member Functions
	InputFileCatalog (fhicl::TableFragment< Config > const &config)

virtual	~InputFileCatalog ()=default

std::size_t	size () const noexcept

FileCatalogItem const &	currentFile () const

size_t	currentIndex () const noexcept

bool	getNextFile (int attempts=5)

bool	hasNextFile (int attempts=5)

void	rewind ()

void	rewindTo (size_t index)

bool	isSearchable () const noexcept

bool	empty () const noexcept

void	finish ()

std::vector< std::string > const &	fileSources () const noexcept

void	commitCatalog ()

std::string &	url () noexcept

std::string const &	url () const noexcept

void	setActive () noexcept

bool	active () const noexcept

Static Public Member Functions
static bool	isPhysical (std::string const &name) noexcept

Static Public Attributes
static constexpr size_t	indexEnd {std::numeric_limits<size_t>::max()}

Private Member Functions
bool	retrieveNextFile (FileCatalogItem &item, int attempts, bool transferOnly=false)

FileCatalogStatus	retrieveNextFileFromCacheOrService (FileCatalogItem &item)

FileCatalogStatus	transferNextFile (FileCatalogItem &item)

Private Attributes
std::vector< std::string >	fileSources_

std::vector< FileCatalogItem >	fileCatalogItems_ {{}}

FileCatalogItem	nextItem_ {}

size_t	fileIdx_ {indexEnd}

size_t	maxIdx_ {}

bool	searchable_ {false}

bool	nextFileProbed_ {false}

bool	hasNextFile_ {false}

ServiceHandle< CatalogInterface >	ci_

ServiceHandle< FileTransfer >	ft_

Detailed Description

Definition at line 30 of file InputFileCatalog.h.

Constructor & Destructor Documentation

art::InputFileCatalog::InputFileCatalog ( fhicl::TableFragment< Config > const & config )

explicit

Definition at line 20 of file InputFileCatalog.cc.

References art::errors::CatalogServiceError, ci_, fileCatalogItems_, fileSources_, and searchable_.

     : fileSources_{config().namesParameter()}
   {
 
     if (fileSources_.empty()) {
       throw Exception(errors::CatalogServiceError,
                       "InputFileCatalog::InputFileCatalog()\n")
         << "Empty '" << config().namesParameter.name()
         << "' parameter specified for input source.\n";
     }
 
     // Configure FileDelivery service
     ci_->configure(fileSources_);
     searchable_ = ci_->isSearchable();
 
     if (searchable_)
       fileCatalogItems_.resize(fileSources_.size());
   }

virtual art::InputFileCatalog::~InputFileCatalog ( )

virtualdefault

Member Function Documentation

bool art::FileCatalog::active ( ) const

inlinenoexceptinherited

Definition at line 113 of file FileCatalog.h.

     {
       return active_;
     }

void art::FileCatalog::commitCatalog ( )

inlineinherited

Definition at line 90 of file FileCatalog.h.

91 {}

FileCatalogItem const & art::InputFileCatalog::currentFile ( ) const

Definition at line 41 of file InputFileCatalog.cc.

References fileCatalogItems_, fileIdx_, indexEnd, art::errors::LogicError, and maxIdx_.

Referenced by finish().

   {
     if (fileIdx_ == indexEnd) {
       throw Exception(
         errors::LogicError,
         "Cannot access the current file while the file catalog is empty!");
     }
     assert(fileIdx_ <= maxIdx_);
     return fileCatalogItems_[fileIdx_];
   }

size_t art::InputFileCatalog::currentIndex ( ) const

noexcept

Definition at line 53 of file InputFileCatalog.cc.

References fileIdx_.

   {
     return fileIdx_;
   }

bool art::InputFileCatalog::empty ( ) const

inlinenoexcept

Definition at line 55 of file InputFileCatalog.h.

     {
       return fileCatalogItems_.empty();
     }

std::vector<std::string> const& art::InputFileCatalog::fileSources ( ) const

inlinenoexcept

Definition at line 61 of file InputFileCatalog.h.

     {
       return fileSources_;
     }

void art::InputFileCatalog::finish ( )

Definition at line 242 of file InputFileCatalog.cc.

References ci_, art::CONSUMED, currentFile(), fileCatalogItems_, fileIdx_, and indexEnd.

Referenced by retrieveNextFile().

   {
     if (fileIdx_ != indexEnd          // there is a current file
         && !currentFile().skipped()   // not skipped
         && !currentFile().consumed()) // not consumed
     {
       ci_->updateStatus(currentFile().uri(), FileDisposition::CONSUMED);
       fileCatalogItems_[fileIdx_].consume();
     }
   }

bool art::InputFileCatalog::getNextFile ( int attempts = 5 )

Definition at line 59 of file InputFileCatalog.cc.

References fileCatalogItems_, fileIdx_, hasNextFile_, indexEnd, maxIdx_, nextFileProbed_, nextItem_, retrieveNextFile(), and searchable_.

   {
     // get next file from FileDelivery service and give it to
     // currentFile_ object returns false if theres no more file
     //
     // If hasNextFile() has been called prior to getNextFile(), it
     // does not actually go fetch the next file from FileDelivery
     // service, instead, it advances the iterator by one and make the
     // "hidden" next file current.
 
     if (nextFileProbed_ && !hasNextFile_)
       return false;
 
     if ((nextFileProbed_ && hasNextFile_) ||
         retrieveNextFile(nextItem_, attempts)) {
       nextFileProbed_ = false;
       fileIdx_ =
         (fileIdx_ == indexEnd) ? 0 : (searchable_ ? (fileIdx_ + 1) : 0);
       if (fileIdx_ > maxIdx_)
         maxIdx_ = fileIdx_;
       fileCatalogItems_[fileIdx_] = nextItem_;
       return true;
     }
 
     return false;
   }

bool art::InputFileCatalog::hasNextFile ( int attempts = 5 )

Definition at line 87 of file InputFileCatalog.cc.

References hasNextFile_, nextFileProbed_, nextItem_, and retrieveNextFile().

   {
     // A probe. It tries (and actually does) retrieve the next file
     // from the FileDelivery service. But does not advance the current
     // file pointer
     if (nextFileProbed_)
       return hasNextFile_;
 
     hasNextFile_ = retrieveNextFile(nextItem_, attempts);
     nextFileProbed_ = true;
 
     return hasNextFile_;
   }

static bool art::FileCatalog::isPhysical ( std::string const & name )

inlinestaticnoexceptinherited

Definition at line 93 of file FileCatalog.h.

Referenced by transferNextFile().

     {
       return (name.empty() || name.find(':') != std::string::npos);
     }

bool art::InputFileCatalog::isSearchable ( ) const

inlinenoexcept

Definition at line 50 of file InputFileCatalog.h.

     {
       return searchable_;
     }

bool art::InputFileCatalog::retrieveNextFile	(	FileCatalogItem &	item,
		int	attempts,
		bool	transferOnly = `false`
	)

private

Definition at line 102 of file InputFileCatalog.cc.

References ci_, art::errors::Configuration, art::DELIVERY_ERROR, finish(), art::NO_MORE_FILES, retrieveNextFileFromCacheOrService(), art::SKIPPED, art::SUCCESS, art::TRANSFER_ERROR, transferNextFile(), art::TRANSFERRED, and art::FileCatalogItem::uri().

Referenced by getNextFile(), and hasNextFile().

   {
     // Tell the service the current opened file (if there is one) is consumed
     finish();
 
     // retrieve (deliver and transfer) next file from service
     // or, do the transfer only
     FileCatalogStatus status;
     if (transferOnly) {
       status = transferNextFile(item);
     } else {
       status = retrieveNextFileFromCacheOrService(item);
     }
 
     switch (status) {
     case FileCatalogStatus::SUCCESS: {
       // mark the file as transferred
       ci_->updateStatus(item.uri(), FileDisposition::TRANSFERRED);
       return true;
     }
     case FileCatalogStatus::NO_MORE_FILES: {
       return false;
     }
     case FileCatalogStatus::DELIVERY_ERROR: {
       if (attempts <= 1) {
         throw Exception(errors::Configuration,
                         "InputFileCatalog::retrieveNextFile()\n")
           << "Delivery error encountered after reaching maximum number of "
              "attemtps!";
       }
       return retrieveNextFile(item, attempts - 1, false);
     }
     case FileCatalogStatus::TRANSFER_ERROR: {
       if (attempts <= 1) {
         // if we end up with a transfer error, the method returns
         // with a true flag and empty filename. Weired enough, but
         // the next file does exist we just cannot retrieve it. Therefore
         // we notify the service that the file has been skipped
         ci_->updateStatus(item.uri(), FileDisposition::SKIPPED);
         return true;
       }
       return retrieveNextFile(item, attempts - 1, true);
     }
     }
     assert(false);
     return false; // Unreachable, but keeps the compiler happy
   }

FileCatalogStatus art::InputFileCatalog::retrieveNextFileFromCacheOrService ( FileCatalogItem & item )

private

Definition at line 153 of file InputFileCatalog.cc.

References ci_, art::DELIVERY_ERROR, fileCatalogItems_, fileIdx_, maxIdx_, art::NO_MORE_FILES, art::SUCCESS, and transferNextFile().

Referenced by retrieveNextFile().

   {
     // Try to get it from cached files
     if (fileIdx_ < maxIdx_) {
       item = fileCatalogItems_[fileIdx_ + 1];
       return FileCatalogStatus::SUCCESS;
     }
 
     // Try to get it from the service
     std::string uri;
     double wait = 0.0;
 
     // get file delivered
     int const result = ci_->getNextFileURI(uri, wait);
 
     if (result == FileDeliveryStatus::NO_MORE_FILES)
       return FileCatalogStatus::NO_MORE_FILES;
 
     if (result != FileDeliveryStatus::SUCCESS)
       return FileCatalogStatus::DELIVERY_ERROR;
 
     item = FileCatalogItem("", "", uri);
 
     // get file transfered
     return transferNextFile(item);
   }

void art::InputFileCatalog::rewind ( )

Definition at line 213 of file InputFileCatalog.cc.

References fileIdx_, art::errors::LogicError, and searchable_.

   {
     if (!searchable_) {
       throw Exception(errors::LogicError, "InputFileCatalog::rewind()\n")
         << "A non-searchable catalog is not allowed to rewind!";
     }
     fileIdx_ = 0;
   }

void art::InputFileCatalog::rewindTo ( size_t index )

Definition at line 223 of file InputFileCatalog.cc.

References fileIdx_, art::errors::InvalidNumber, art::errors::LogicError, maxIdx_, and searchable_.

   {
     // rewind to a previous file location in the catalog service is
     // not rewinded. only usable when
     // FileDeliveryService::areFilesPersistent() is true
     if (!searchable_) {
       throw Exception(errors::LogicError, "InputFileCatalog::rewindTo()\n")
         << "A non-searchable catalog is not allowed to rewind!";
     }
 
     if (index > maxIdx_) {
       throw Exception(errors::InvalidNumber, "InputFileCatalog::rewindTo()\n")
         << "Index " << index << " is out of range!";
     }
 
     fileIdx_ = index;
   }

void art::FileCatalog::setActive ( )

inlinenoexceptinherited

Definition at line 108 of file FileCatalog.h.

     {
       active_ = true;
     }

std::size_t art::InputFileCatalog::size ( void ) const

inlinenoexcept

Definition at line 39 of file InputFileCatalog.h.

     {
       return fileCatalogItems_.size();
     }

FileCatalogStatus art::InputFileCatalog::transferNextFile ( FileCatalogItem & item )

private

Definition at line 181 of file InputFileCatalog.cc.

References art::errors::Configuration, art::FileCatalogItem::fileName(), ft_, art::FileCatalog::isPhysical(), art::FileCatalogItem::logicalFileName(), art::FileCatalogItem::skip(), art::SUCCESS, art::TRANSFER_ERROR, and art::FileCatalogItem::uri().

Referenced by retrieveNextFile(), and retrieveNextFileFromCacheOrService().

   {
     std::string pfn;
     int const result = ft_->translateToLocalFilename(item.uri(), pfn);
 
     if (result != FileTransferStatus::SUCCESS) {
       item.fileName("");
       item.logicalFileName("");
       item.skip();
       return FileCatalogStatus::TRANSFER_ERROR;
     }
 
     // successfully retrieved the file
     std::string lfn = pfn;
     boost::trim(pfn);
 
     if (pfn.empty()) {
       throw Exception(errors::Configuration,
                       "InputFileCatalog::retrieveNextFileFromCacheService()\n")
         << "An empty string specified in parameter for input source.\n";
     }
 
     if (isPhysical(pfn)) {
       lfn.clear();
     }
 
     item.fileName(pfn);
     item.logicalFileName(lfn);
     return FileCatalogStatus::SUCCESS;
   }