DybArchiveOutput.cc

Go to the documentation of this file.

#include "DybArchiveOutput.h"

#include "DybKernel/EventStoreIncident.h"
#include "Event/HeaderObject.h"

#include "GaudiKernel/DataSvc.h"
#include "GaudiKernel/IIncidentSvc.h"

#include <sstream>

using namespace DayaBay;

DybArchiveOutput::DybArchiveOutput (const std::string& name, ISvcLocator* pSvcLocator)
 : GaudiAlgorithm(name, pSvcLocator)
 , m_aes(0)
 , m_store(0)
 , m_incsvc(0)
{
    declareProperty("WindowSeconds", m_window = 0.0, "Time window to apply to the AES");    
    declareProperty("SaveOutput", m_saveFlag = false, "Set to true to save ouput, default is false");
    declareProperty("RegSeqLocation",
                    m_regSeqLocation = DayaBay::RegistrationSequenceLocation::Default,
                    "Location to RegistrationSequence");
    declareProperty("ArchiveSvc",
                    m_aesName = "EventDataArchiveSvc",
                    "Name of Archive Event Store");
    declareProperty("DybStorageSvc",
                    m_dybStorageSvcName = "DybStorageSvc",
                    "Name of an IDybStorageSvc");
}

StatusCode DybArchiveOutput::initialize()
{
    StatusCode sc = GaudiAlgorithm::initialize();
    if (sc.isFailure()) return sc;

    sc = service(m_aesName,m_aes);
    if (sc.isFailure()) return sc;

    if (m_saveFlag) {
        sc = service(m_dybStorageSvcName, m_store);
        if (sc.isFailure()) return sc;    
    }

    sc = service("IncidentSvc",m_incsvc,true);
    if (sc.isFailure()) return sc;

    debug() << "initialized" << endreq;

    return StatusCode::SUCCESS;
}

static std::string dostr(const DataObject* obj)
{
    std::stringstream ss;
    ss << obj->name() << ": refCount=" << obj->refCount();
    const TemporalDataObject* tdo = dynamic_cast<const TemporalDataObject*>(obj);
    if (!tdo) {
        return ss.str();
    }
    ss << " time:" << tdo->earliest() << " --> " << tdo->latest();
    const HeaderObject* ho = dynamic_cast<const HeaderObject*>(tdo);
    if (!ho) {
        return ss.str();
    }
    ss << " execNumber=" << ho->execNumber();
    return ss.str();
}

DybArchiveList* DybArchiveOutput::aes_list(const std::string& location)
{
    // return the list at the given location or null if it d.n.e.  use
    // local cache to avoid somewhat expensive calls to
    // aes->retrieveObject().

    AESListMap_t::iterator it = m_aesListMap.find(location);
    if (it != m_aesListMap.end()) { // cache hit
        return it->second;
    }

    DataObject* object = 0;
    StatusCode sc = m_aes->retrieveObject(location,object);
    if (sc.isFailure()) {       // not here now but maybe later.
        return 0;
    }
    DybArchiveList* aesList = dynamic_cast<DybArchiveList*>(object);
    if (!aesList) {
        // fixme: add warning msg
        return 0;
    }
    
    m_aesListMap[location] = aesList;
    return aesList;
}

StatusCode DybArchiveOutput::store(DayaBay::RegistrationSequence* rs)
{
    if (! m_store) {            // not configure to store data
        return StatusCode::SUCCESS;
    }

    debug() << "Storing RegistrationSequence with "
            << rs->size() << " registrations spanning " 
            << rs->earliest() << " --> " << rs->latest()
            << endreq;
    return m_store->store(*rs);
}

DybArchiveList::iterator DybArchiveOutput::find_in_archive(DybArchiveList* aeslist, const DataObject* object)
{
    DybArchiveList::iterator it, done = aeslist->end();
    for (it = aeslist->begin(); it != done; ++it) {
        if (object == *it) {
            return it;
        }
    }
    return done;
}


StatusCode DybArchiveOutput::erase(const DataObject* object, const std::string& path)
{
    DybArchiveList* lst = aes_list(path);
    if (!lst) {
        error() << "No AES list for " << path << endreq;
        return StatusCode::FAILURE;
    }

    DybArchiveList::iterator todie = find_in_archive(lst,object);
    if (todie == lst->end()) {
        error() << "Failed to find object for " << path
                << " object:" << dostr(object)
                << endreq;
        return StatusCode::FAILURE;
    }

    m_incsvc->fireIncident(EventStoreIncident(object, false, "TrimmingAgent"));

    debug() << "Erasing object from " << path
            << " object:" << dostr(object)
            << endreq;

    lst->erase(todie);  // when Registration is dtor'ed, object.release() is called

    return StatusCode::SUCCESS;
}
StatusCode DybArchiveOutput::trim(DayaBay::RegistrationSequence* rs)
{
    const IRegistrationSequence::Registrations& regs = rs->registrations();
    IRegistrationSequence::Registrations::const_iterator regit, regend = regs.end();

    // First remove all registrations from their AES lists
    for (regit = regs.begin(); regit != regend; ++regit) {
        const ObjectReg& reg = (*regit);
        StatusCode sc = erase(reg.object(),reg.path());
        if (sc.isFailure()) {
            return StatusCode::FAILURE;
        }

    }

    // Finally, trim the RegistrationSequence itself.
    StatusCode sc = erase(rs,m_regSeqLocation);
    if (sc.isFailure()) {
        return StatusCode::FAILURE;
    }

    return StatusCode::SUCCESS;
}    


RegistrationSequence* DybArchiveOutput::condemned()
{
    // Try to return the next RS if it is out of the window

    DybArchiveList* rslist = aes_list(m_regSeqLocation);
    if (! rslist) return 0;
    if (rslist->size() <= 1) return 0;

    // newest objects are push_front'ed, so we must check the back for oldest.
    DybArchiveList::reverse_iterator rit_old = rslist->rbegin();
    DybArchiveList::iterator rit_new = rslist->begin();
    RegistrationSequence* rs_old = dynamic_cast<RegistrationSequence*>(*rit_old);
    RegistrationSequence* rs_new = dynamic_cast<RegistrationSequence*>(*rit_new);

    if (!rs_old || !rs_new) {
        error() << "Not a RegistrationSequence at " << m_regSeqLocation << endreq;
        return 0;
    }

    TimeStamp threshold = rs_new->earliest();
    threshold.Add(-1.0 * m_window);

    if (rs_old->earliest() > threshold) {
        return 0;
    }

    return rs_old;              // I salute you.
}

StatusCode DybArchiveOutput::process(RegistrationSequence* rs)
{
    // Process one RS, maybe store and then trim

    StatusCode sc = store(rs);
    if (sc.isFailure()) {
        error() << "Failed to store RegistrationSequence: " << dostr(rs) << endreq;
        return sc;
    }

    sc = trim(rs);
    if (sc.isFailure()) {
        error() << "Failed to trim RegistrationSequence: " << dostr(rs) << endreq;
        return sc;
    }

    return StatusCode::SUCCESS;
 }

StatusCode DybArchiveOutput::execute()
{
    // save and trim until no more 
    while (true) {
        RegistrationSequence* rs = condemned();
        if (!rs) { 
            return StatusCode::SUCCESS;
        }

        StatusCode sc = process(rs);
        if (sc.isFailure()) {
            // fixme: error msg
            return sc;
        }
    }
    
    // actually, should never reach here.
    return StatusCode::SUCCESS;
}

StatusCode DybArchiveOutput::sweep_up()
{
    // process any remaining data left in the AES.

    DybArchiveList* rslist = aes_list(m_regSeqLocation);

    debug() << "sweeping up with " << rslist->size() << " RegistrationSequences to process" << endreq;

    // be careful iterating over this list because it is being erased as we go
    while (true) {              
        if (!rslist->size()) {
            break;
        }

        DybArchiveList::reverse_iterator rit = rslist->rbegin();
        RegistrationSequence* rs = dynamic_cast<RegistrationSequence*>(*rit);
        StatusCode sc = process(rs);
        if (sc.isFailure()) {
            return sc;
        }
    }
    return StatusCode::SUCCESS;
}
StatusCode DybArchiveOutput::stop()
{
    debug() << "stop called" << endreq;

    return GaudiAlgorithm::stop();
}
StatusCode DybArchiveOutput::finalize()
{
    debug() << "finalize" << endreq;

    StatusCode sc = sweep_up();
    if (sc.isFailure()) {
        return sc;
    }

    return GaudiAlgorithm::finalize();
}