DybOutputFileRoller.cc

Go to the documentation of this file.

#include "DybOutputFileRoller.h"
#include "Event/RegistrationSequence.h"

#include "TString.h"            // for Form()

#include <cstdlib>              // for atoi()
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>             // for stat

using namespace std;
using namespace DayaBay;

DybOutputFileRoller::DybOutputFileRoller(const std::string& name, 
                                         ISvcLocator* svcloc)
    : GaudiAlgorithm(name,svcloc)
    , m_current_execnum(0)
    , m_rio(0)
{
    declareProperty("Condition",m_condition="",
                    "Set the default condition for rolling to a new output file");

    // This condition map will be a trivial one just containing
    // "default" as set by the global condition above.  It can be made
    // non-trivial when/if RootIO can handle reading input streams
    // with non-common boundaries.  See #671.

    //declareProperty("ConditionMap",m_condpropmap,
    //                "Set the condition map for rolling each stream to a new output file");
}

DybOutputFileRoller::~DybOutputFileRoller()
{
}    

StatusCode DybOutputFileRoller::initialize()
{
    StatusCode sc = this->GaudiAlgorithm::initialize();
    if (sc.isFailure()) return sc;

    // Get IRootIOSvc
    sc = this->service("RootIOCnvSvc",m_rio);
    if (sc.isFailure()) return sc;

    m_filemap = m_rio->fileMap();
    if (!m_filemap.size()) {
        return StatusCode::SUCCESS; // we just won't play this time
    }
    
    // munge the output filenames where needed
    {
        IRootIOSvc::FileMap::iterator it, done = m_filemap.end();
        for (it = m_filemap.begin(); it != done; ++it) {
            string fname = Form(it->second.c_str(),0);
            if (fname != it->second) {
                continue;           // already has a %d code that does something
            }
            
            string::size_type ind = it->second.rfind(".");
            // shouldn't happen, but whatevs, we'll roll with it
            if (string::npos == ind) { 
                ind = it->second.size(); 
            }
            
            // replace filename with one that has a %d code.
            fname = it->second.substr(0,ind) + "_%04d" + it->second.substr(ind);
            it->second = fname;
        }
    }
    
    // specifically set default condition if one is given
    if (m_condition.size()) {
        m_condpropmap["default"] = m_condition;
    }

    { // parse conditions map and strip out any explicit conditions
      // with no matching entry in output file map
        ConditionPropertyMap_t::iterator it, done = m_condpropmap.end();
        for (it=m_condpropmap.begin(); it != done; ++it) {
            string stream = it->first;
            string condition = it->second;

            if (stream != "default") {
                IRootIOSvc::FileMap::iterator fit = m_filemap.find(stream);
                if (fit == m_filemap.end()) {
                    error() << "Condition \"" << condition 
                            << "\" for stream \"" << stream 
                            << "\" has no output file.  " 
                            << "An entry can be specified with nuwa.py's"
                            << " -O/--output-stream option"
                            << endreq;
                    return StatusCode::FAILURE;
                }
            }
            
            string::size_type clen = condition.size();
            if (!clen) {
                error() << "Given a null condition for stream \""
                        << stream << "\"" << endreq;
                return StatusCode::FAILURE;
            }
            int size = atoi(condition.substr(0,clen-1).c_str());
            char unit = condition[clen-1];
            ConditionData cd = ConditionData(size,unit);
            cd.cycles = size;   // special init, count down on this if an E condition
            m_cdmap[stream] = cd;
            info() << "Condition for stream " << stream << ": " << size << unit << endreq;
            sc = this->roll(stream,m_cdmap[stream]);
            if (sc.isFailure()) return sc;
        }
    }

    return StatusCode::SUCCESS;
}
StatusCode DybOutputFileRoller::roll(std::string stream, ConditionData& cd)
{
    ++cd.count;

    IRootIOSvc::FileMap::iterator fit = m_filemap.find(stream);
    if (fit == m_filemap.end()) {
        error() << "Failed to find output filespec for \"" << stream << "\"" << endreq;
        return StatusCode::FAILURE;
    }

    string fname = Form(fit->second.c_str(),cd.count);
    m_rio->associateOutput(fname.c_str(),stream.c_str());

    info() << "Rolled " << stream << " to file " << fname << endreq;

    return StatusCode::SUCCESS;
}

StatusCode DybOutputFileRoller::check_m(std::string stream, ConditionData& cd)
{
    IRootIOSvc::FileMap fm = m_rio->fileMap();
    IRootIOSvc::FileMap::iterator fit = fm.find(stream);
    if (fit == fm.end()) {
        error() << "Failed to find an output file for stream \"" << stream << "\""
                << endreq;
        return StatusCode::FAILURE;
    }
    string fname = fit->second;
    struct stat si;
    int err = stat(fname.c_str(), &si);
    if (err) {
        warning() << "Failed to stat " << fname
                  << ".  Maybe no output yet?  Letting it go."
                  << endreq;
        return StatusCode::SUCCESS;
    }
        
    if (si.st_size/1000000 < cd.size) {  // bytes to mibibytes
        return StatusCode::SUCCESS;
    }
    info () << "Reached " << cd.size << " MiB size" << endreq;
    return this->roll(stream,cd);
}


// check based on execution cycle count
StatusCode DybOutputFileRoller::check_e(std::string stream, ConditionData& cd)
{
    info() << "check_e " << stream << " " << cd.cycles << " " << (void*)&cd << endreq;

    // check for first time
    if (cd.cycles < 0) {
        cd.cycles = cd.size;
    }

    if (cd.cycles>0) {
        --cd.cycles;
        return StatusCode::SUCCESS;
    }
    cd.cycles = cd.size-1;        // reset countdown
    info () << "Reached " << cd.size << " execution cycles" << endreq;
    return this->roll(stream,cd);
}

// for this to work, the algorithm must be placed after something that
// loads data into the TES.
StatusCode DybOutputFileRoller::check_s(std::string stream, ConditionData& cd)
{
    RegistrationSequence* rs = 
        get<RegistrationSequence>(RegistrationSequence::defaultLocation());

    TimeStamp now = rs->earliest();

    // check for first time
    if (cd.timestamp == TimeStamp::GetBOT()) {
        cd.timestamp = now;
        return StatusCode::SUCCESS;
    }

    TimeStamp dt = now - cd.timestamp;
    int deltat = (int)((double)dt);
    if (deltat < cd.size) {     // not yet
        return StatusCode::SUCCESS;
    }
    
    cd.timestamp = now;
    info () << "Reached " << cd.size << " seconds at " << now << endreq;
    return this->roll(stream,cd);
}

StatusCode DybOutputFileRoller::execute()
{
    if (!m_filemap.size()) {    // no output files, bail blissfully
        return StatusCode::SUCCESS;
    }

    RegistrationSequence* rs = get<RegistrationSequence>(RegistrationSequence::defaultLocation());
    TimeStamp now = rs->earliest();
    ++m_current_execnum;
    info() << "Execution #" << m_current_execnum << " @ " << now << endreq;

    CDMap_t::iterator it, done = m_cdmap.end();
    for (it=m_cdmap.begin(); it != done; ++it) {
        string stream = it->first;
        ConditionData& cd = it->second;

        StatusCode sc;

        switch (cd.unit) {
        case 'M':
            sc = check_m(stream,cd);
            break;
        case 'E':
            sc = check_e(stream,cd);
            break;
        case 'S':
            sc = check_s(stream,cd);
            break;
        default:
            error() << "Unknown condition unit '" << cd.unit << "'" 
                    << " for stream \"" << stream << "\""
                    << endreq;
            sc = StatusCode::FAILURE;
        }
        if (sc.isFailure()) return sc;
    }
    return StatusCode::SUCCESS;
}

StatusCode DybOutputFileRoller::finalize()
{
    return this->GaudiAlgorithm::finalize();
}