event_store.cpp

Go to the documentation of this file.

#include "stdafx.h"
#include "event_store.h"
#include "dae_events.h"
#include "icputils.h"
#include "isisraw.h"

EventStore::EventStore() : Base(), m_mempool(m_mempool_size)
{
        setLoggerName("EventStore");
        LOGSTR_DEBUG("Constructing event store");
        m_threads.addCapacity(50);
}

uint64_t EventStore::numRawEvents()
{
        uint64_t n = 0;
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                n += ec.card->numRawEvents();
        }
        return n;
}

uint64_t EventStore::nNewEventWords(DAEstatus& status)
{
        uint64_t n = 0;
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                n += ec.card->nNewEventWords(status);
        }
        return n;
}

void EventStore::getEventSourceIDs(std::vector<int>& ids)
{
        ids.clear();
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                ids.push_back(ec.card->eventSourceID());
        }
}

void EventStore::getEnabledEventSourceIDs(std::vector<int>& ids)
{
        ids.clear();
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                if (ec.enable)
                {
                        ids.push_back(ec.card->eventSourceID());
                }
        }
}

uint64_t EventStore::numGoodEvents()
{
        uint64_t n = 0;
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                n += ec.card->numGoodEvents();
        }
        return n;
}

void EventStore::findFileInputSources(int run_number, std::set<std::string>& files)
{
        files.clear();
        std::set<std::string> files1;
        Poco::Path search_expr(EventStore::baseDir(), "dc_*.dat");
        search_expr.pushDirectory(Poco::format("run_%d", run_number));
        Poco::Glob::glob(search_expr, files1, Poco::Glob::GLOB_CASELESS); // poco glob is shell like not RE like
        BOOST_FOREACH(const std::string& s, files1)  // not std::string& as we may erase it
        {
                if (s.back() == '/' || s.back() == '\\')   // remove a trailing "/" which is present if we are using SplitFile
                {
                        files.insert(s.substr(0,s.length()-1));
                }
                else
                {
                        files.insert(s);
                }
        }
}

int EventStore::eventSourceIDFromFileInputSource(const std::string& file)
{
        Poco::RegularExpression re("dc_([0-9]+).dat", Poco::RegularExpression::RE_CASELESS);
        Poco::RegularExpression::MatchVec matches;
        std::string s;
        re.match(file, 0, matches);
        if (matches.size() == 2)
        {
                s = file.substr(matches[1].offset, matches[1].length);
                return Poco::NumberParser::parse(s);
        }
        else
        {
                return -1;
        }
}
        
void EventStore::findUnusedFileInputSources(int run_number, std::set<std::string>& files)
{
        findFileInputSources(run_number, files);
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                for(std::set<std::string>::const_iterator it = files.begin(); it != files.end(); ++it)
                {
                        if ( !stricmp(it->c_str(), ec.card->getFileName().c_str()) )
                        {
                                files.erase(it);
                                break;
                        }
                }
        }
}

void EventStore::addFileInputSource(int run_number, FileEventSourceInfo& info)
{
        LOGSTR_DEBUG("Adding file sources");
        std::set<std::string> files;
        info.zero();
        findFileInputSources(run_number, files);
        if (files.size() == 0)
        {
                LOGSTR_WARNING("No event files found for run " << run_number);
        }
        FileEventSourceInfo card_info;
        BOOST_FOREACH(const std::string& f, files)
        {
                int ev_id = eventSourceIDFromFileInputSource(f);
                if (ev_id != -1)
                {
                        FileEventSource* fev = new FileEventSource(run_number, ev_id);
                        m_sources.push_back(fev);
                        fev->readInfo(card_info);
                        info.add(card_info);
                        addCard(fev, (card_info.raw_frames > 0 ? true : false) );
                }
        }
}

void EventStore::addRawFileInputSource(ISISRAW* iraw, int source_id, time_t run_start)
{
        LOGSTR_DEBUG("Adding raw file sources");
        ISISRAWEventSource* fev = new ISISRAWEventSource(iraw, source_id, run_start);
        m_sources.push_back(fev);
        addCard(fev, true);
}

void EventStore::addCard(DAEEventSource* det_card, bool enable)
{
        LOGSTR_DEBUG("Adding source id " << det_card->eventSourceID() << " as " << (enable ? "ENABLED" : "DISABLED") );
        EventStoreCard* ec = new EventStoreCard(this, det_card);
        m_stores.push_back(EventStoreCardHolder(ec, enable));
}

void EventStore::addCards(const std::vector<DAEEventSource*>& det_list) 
{
        LOGSTR_DEBUG("Adding sources");
        BOOST_FOREACH(DAEEventSource* dc, det_list)
        {
                addCard(dc);
        }
}

void EventStore::setCardStatus(bool enable)
{
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                setCardStatus(ec.card->eventSourceID(), enable);
        }
}

void EventStore::setCardStatus(int id, bool enable)
{
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                if (ec.card->eventSourceID() == id)
                {
                        ec.enable = enable;
                }
        }
}

void EventStore::invokeNoFrameCallbacks(NoFrameCallbackMode mode)
{
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                if (ec.enable)
                {
                        ec.card->invokeNoFrameCallbacks(mode);
                }
        }
}

void EventStore::start()
{
        LOGSTR_DEBUG("Starting event store");
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                if (ec.enable)
                {
                        m_threads.start(*(ec.card), Poco::format("EventStoreCard %d", ec.card->eventSourceID()));
                }
        }
}

void EventStore::setOutputFileName(int run_number, bool new_run, DAEstatus& status)
{
        ICPTimer timer("EventStore::setOutputFileName", status);
        LOGSTR_DEBUG("Setting output event file names for run number " << run_number);
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                ec.card->setOutputFileName(run_number, new_run, status);
        }
        timer.info("After setting of card sources", status); 
        // now remove and old files we are not using
        std::set<std::string> files;
        findUnusedFileInputSources(run_number, files);
        BOOST_FOREACH(const std::string& file, files)
        {
                LOGSTR_DEBUG("Removing no longer used source " << file);
                SplitFile::removeFile(file);
        }
}

void EventStore::clearOutputFile(DAEstatus& status)
{
        ICPTimer timer("EventStore::clearOutputFile", status);
        LOGSTR_DEBUG("Clearing output files");
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                ec.card->clearOutputFile(status);
        }
}

void EventStore::closeAndDeleteOutputFile(int run_number, DAEstatus& status)
{
        stop(true);
        closeOutputFile(status);
        deleteOutputFile(run_number, status);
}

void EventStore::closeOutputFile(DAEstatus& status)
{
        stop(false);
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                ec.card->closeFiles();
        }
}

void EventStore::deleteOutputFile(int run_number, DAEstatus& status)
{
        // we loop round files in case some are in use, so hopefully get most of them to save disk space
        std::set<std::string> files;
        findFileInputSources(run_number, files);
        BOOST_FOREACH(const std::string& file, files)
        {
                SplitFile::removeFile(file);
        }
        // clean up anything else
        Poco::File run_dir(baseRunDir(run_number));
        if (run_dir.exists())
        {
                try
                {
                        run_dir.remove(true);
                }
                catch(const std::exception& ex)
                {
                        status.addInfoVa(FAC_DAE, "Unable to remove all files in event directory of run %d: %s", run_number, ex.what());
                }
        }
}

void EventStore::stop(bool immediate)
{
        LOGSTR_DEBUG("Stopping event store");
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                ec.card->stop(immediate);   // request cards to stop when finished current work
        }
        m_threads.joinAll();  // wait for above to happen 
}

bool EventStore::endHeaderSeen()
{
        bool end_seen = true;
        unsigned nseen = 0, nchecked = 0;
//      LOGSTR_TRACE("Checking for end header");
        BOOST_FOREACH(const EventStoreCardHolder& ec, m_stores)
        {
                if (ec.enable)
                {
                        ++nchecked;
                        if ( ec.card->endHeaderSeen() )
                        {
                                ++nseen;
                        }
                        else
                        {
                                end_seen = false;
                        }
                }
        }
        if (nseen > 0)
        {
                LOGSTR_DEBUG("Checking for end header - seen " << nseen << "/" << nchecked);
        }
        return end_seen;
}

template<typename SLOT>
std::list<boost::signals2::connection> EventStore::addCallback(boost::signals2::connection (EventStoreCard::*func)(const SLOT&), const SLOT& slot) 
{
        std::list<boost::signals2::connection> conn_list;
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                conn_list.push_back((ec.card->*func)(slot));
        }
        return conn_list;
}

std::list<boost::signals2::connection> EventStore::addGoodFrameCallback(const FrameCallbackSlotType& slot)
{
        LOGSTR_DEBUG("Adding GOOD frame callback");
        return addCallback(&EventStoreCard::addGoodFrameCallback, slot);
}

std::list<boost::signals2::connection> EventStore::addAllFrameCallback(const FrameCallbackSlotType& slot)
{
        LOGSTR_DEBUG("Adding ALL frame callback");
        return addCallback(&EventStoreCard::addAllFrameCallback, slot);
}

std::list<boost::signals2::connection> EventStore::addNoFrameCallback(const NoFrameCallbackSlotType& slot)
{
        LOGSTR_DEBUG("Adding NO frame callback");
        return addCallback(&EventStoreCard::addNoFrameCallback, slot);
}

std::list<boost::signals2::connection> EventStore::addGoodEventCallback(const EventCallbackSlotType& slot)
{
        LOGSTR_DEBUG("Adding GOOD event callback");
        return addCallback(&EventStoreCard::addGoodEventCallback, slot);
}

std::list<boost::signals2::connection> EventStore::addAllEventCallback(const EventCallbackSlotType& slot) 
{
        LOGSTR_DEBUG("Adding ALL event callback");
        return addCallback(&EventStoreCard::addAllEventCallback, slot);
}

EventStore::~EventStore()
{
        LOGSTR_DEBUG("Destructor");
        stop(false);
        clear();
}

void EventStore::clear()
{
        clearAllCallbacks();
        clearSources();
        clearStores();
}

void EventStore::clearSources()
{
        BOOST_FOREACH(FileEventSourceBase* es, m_sources)
        {
                delete es;
        }
        m_sources.clear();
}

void EventStore::clearStores()
{
        BOOST_FOREACH(EventStoreCardHolder& ec, m_stores)
        {
                delete ec.card;
        }
        m_stores.clear();
}


const int EventStore::m_mempool_size = 32 * 1024 * 1024;

isisU32_t* EventStore::getBuffer()
{
        return reinterpret_cast<isisU32_t*>(m_mempool.get());
}

void EventStore::releaseBuffer(isisU32_t* buffer)
{
        m_mempool.release(buffer);
}

#define THROW_RETURN_NE(__comm, __val, __mess) \
        if ( (__comm) != (__val) ) \
        { \
            throw std::runtime_error(__mess); \
        }

EventStoreCard::EventStoreCard(EventStore* es, DAEEventSource* dc) : Base(), m_store(es), m_dc(dc), 
        m_run_number(-1), m_keep_going(true), m_immediate_stop(false), m_len_buffer(0), m_buffer(NULL)
{
        m_data.newFile();
        setLoggerName("EventStoreCard");
        LOGSTR_DEBUG("Constructing event store card for source id " << m_dc->eventSourceID());
}


std::string EventStoreCard::fileNameForDC(int run_number, int dc_number)
{
        std::string file_name;
        Poco::format(file_name, "dc_%d.dat", dc_number);
        Poco::Path p(EventStore::baseDir(), file_name);
        p.pushDirectory(Poco::format("run_%d", run_number));
        return p.toString();
}

std::string EventStoreCard::infoFileNameForDC(int run_number, int dc_number)
{
        std::string info_file_name;
        Poco::format(info_file_name, "dc_%d.info", dc_number);
        Poco::Path p(EventStore::baseDir(), info_file_name);
        p.pushDirectory(Poco::format("run_%d", run_number));
        return p.toString();
}

void EventStoreCard::getInfoFileData(const std::string& file_name, InfoFileData& info)
{
        if ( atomic_read(file_name, ".old", &info, 1) != 0 )
        {
                info.newFile();
        }
// need to check the real event file size not the info file size
//      else if ( (file_size_bytes(file_name) / sizeof(uint32_t)) < info.total )
//      {
//              info.total = 0;
//              LOGSTR_WARNING("File " << file_name << " is short ");
//      }
}

void EventStoreCard::clearOutputFile(DAEstatus& status)
{
        if (m_keep_going)
        {
                LOGSTR_WARNING("Clearing output file while running");
        }
        LOGSTR_DEBUG("Clearing output for source id " << m_dc->eventSourceID());
        m_file.seek(0, SEEK_SET);
        m_file.discardContents();
        m_data.newFile();
        updateFilePosition(status);
}


void EventStoreCard::setOutputFileName(int run_number, bool new_run, DAEstatus& status)
{
        LOGSTR_DEBUG("Setting output run number to " << run_number);
        if (m_run_number == run_number && m_file.isOpen())  // is it already open?
        {
                LOGSTR_DEBUG("file already open");
        }
        else
        {
                m_run_number = run_number;
                std::string file_name, info_file_name; 
                m_file_name = fileNameForDC(run_number, m_dc->eventSourceID());
                Poco::Path dir_name(m_file_name);
                dir_name.setFileName("");
                Poco::File(dir_name).createDirectories();
                m_info_file_name = infoFileNameForDC(run_number, m_dc->eventSourceID());
                LOGSTR_DEBUG("Using event file path " << m_file_name);
                if (m_file.isOpen())
                {
                        m_file.close();
                }
                m_file.open(m_file_name.c_str(), IFile::ReadWriteCreate, _SH_DENYWR, true);
                if (!m_file.isOpen())
                {
                        throw std::runtime_error("fopen");
                }
                getInfoFileData(m_info_file_name, m_data); 
        }
        isisU32_t last_address = m_dc->getLastAddressRead();
        if (new_run)
        {
                 clearOutputFile(status);
                 m_data.last_address_read = last_address;
        }
        else
        {
                if (last_address != m_data.last_address_read)
                {
                        LOGSTR_WARNING("Mismatch in last_address_read for source id " << m_dc->eventSourceID() << ": DAE 0x" << std::hex << last_address << " != ICP 0x" << m_data.last_address_read << std::dec);
                }
                // if is possible the program may have been terminated during updateFilePosition() after writing
                // last_address to the info file but before calling updateDAEEventLastRead()
                // this is to sync the info file to the dae
                m_dc->setLastAddressRead(m_data.last_address_read);
        }
        m_dc->updateDAEEventLastRead(status);
        if (m_file.seek(m_data.file_words * 4, SEEK_SET) != 0)
        {
                LOGSTR_ERROR("Seek failed for " << m_info_file_name);
                m_data.newFile();
        }
        LOGSTR_INFORMATION("Using info file " << m_info_file_name << " words " << m_data.file_words << " LF=" << m_data.last_frame << " GF=" << m_data.good_frames);
        updateFilePosition(status);
}

void EventStoreCard::updateFilePosition(DAEstatus& status)
{
        if (m_file.isOpen())
        {
                m_data.last_address_read = m_dc->getLastAddressRead();
                LOGSTR_DEBUG("Updating file position to " << m_data.file_words << " for " << m_file_name);
                if (atomic_write(m_info_file_name, ".old", ".new", &m_data, 1) != 0)
                {
                        throw std::runtime_error("atomic_write");
                }
                m_dc->updateDAEEventLastRead(status);
        }
}

Poco::Path EventStore::m_base_dir;

void EventStoreCard::run()
{
        DAEstatus status;
        uint64_t num_unread;
        m_keep_going = true;
        m_immediate_stop = false;
        LOGSTR_DEBUG("Starting event thread for " << m_file_name);
        while(m_keep_going)
        {
                m_buffer = m_store->getBuffer();
                m_len_buffer = m_dc->getNewEvents(m_buffer, m_store->bufferSize(), num_unread, m_data.end_header_present, status);
                if (num_unread > 0)
                {
                        LOGSTR_DEBUG("words remaining " << num_unread);
                }
                if (m_len_buffer > 0)
                {
                        Poco::RunnableAdapter<EventStoreCard> histogram_events(*this, &EventStoreCard::checkAndHistogramEvents);
                        m_threads.start(histogram_events, "histogram_events");
                        flushEvents(status);
                        m_threads.joinAll();
                }
                else
                {
                        invokeNoFrameCallbacks(NoFrameRunning);
//                      LOGSTR_DEBUG("No data - end present " << m_end_header_present);
                }
                updateFilePosition(status);  // need to do this even as no data so end_header_present is written
                m_store->releaseBuffer(m_buffer);
                if (m_len_buffer == 0)
                {
                        Poco::Thread::sleep( m_data.end_header_present ? 2000 : 200 );
                }
                Poco::Thread::yield();
        }
        num_unread = m_dc->nNewEventWords(status);
        if (num_unread > 0)
        {
                LOGSTR_WARNING("Terminating event thread for " << m_file_name << " with " << num_unread << " event words remaining");
        }
        invokeNoFrameCallbacks(NoFrameStop);   // to indicate end of data stream
}

uint64_t EventStoreCard::nNewEventWords(DAEstatus& status)
{
        return m_dc->nNewEventWords(status);
}

void EventStoreCard::invokeNoFrameCallbacks(NoFrameCallbackMode mode)
{
        m_no_frame_callbacks(m_data.end_header_present, mode, m_dc->eventSourceID());   // to indicate end of data stream
}

void EventStoreCard::flushEvents(DAEstatus& status)
{
        if (m_file.isOpen())
        {
                LOGSTR_DEBUG("Flushing " << m_len_buffer << " words to " << m_file_name);
                if (m_file.write(m_buffer, sizeof(isisU32_t), m_len_buffer) != m_len_buffer)
                {
                        throw std::runtime_error("fwrite");
                }
                if (m_file.flush() != 0)
                {
                        throw std::runtime_error("flush");
                }
                m_data.file_words += m_len_buffer;
        }
}

bool EventStoreCard::fixFrameHeader(DAEEventHeader& head)
{
        int fnum = m_data.last_frame + 1;
        if ( head.frame_number != fnum )
        {
                head.frame_number = fnum;
                return true;
        }
        else
        {
                return false;
        }
}

void EventStoreCard::fixBadFrameHeader(DAEEventHeader& head)
{
        head.info.bad_frame = 0;
        head.period = 0;
        // head.protons , head.time ?
}

void EventStoreCard::checkAndHistogramEvents()
{
        LOGSTR_DEBUG("Checking events for " << m_file_name << " num_words = " << m_len_buffer);
        DAEstatus status;
        static bool frameworkaround_default = Poco::Util::Application::instance().config().getBool("isisicp.frameworkaround", false);
//      static const std::string workaround_string(Poco::format("isisicp.frameworkaround.card%d", m_dc->eventSourceID()));
        static bool frameworkaround = frameworkaround_default; // Poco::Util::Application::instance().config().getBool(workaround_string, frameworkaround_default);
        static bool badheaderworkaround = Poco::Util::Application::instance().config().getBool("isisicp.badheaderworkaround", false);
        unsigned n;
        int i, num_words = 0;  // num words of buffer read so far
        std::string error_message;
        unsigned num_frames_fixed = 0;
        while(!m_immediate_stop && num_words < m_len_buffer)
        {
                if (m_data.header_words < DAE_EVENT_HEADER_WORDS)  // incomplete or no current header
                {
                        n = std::min(DAE_EVENT_HEADER_WORDS - m_data.header_words, m_len_buffer - num_words);
                        memcpy(reinterpret_cast<uint32_t*>(&m_data.header) + m_data.header_words, m_buffer + num_words, n * sizeof(uint32_t));
                        num_words += n;
                        m_data.header_words += n;
                        m_data.num_events = 0;
                }
                if (m_data.header_words != DAE_EVENT_HEADER_WORDS)
                {
                        LOGSTR_DEBUG("incomplete header " << m_data.header_words << " - waiting for more data");
                        break;
                }
                if ( frameworkaround && fixFrameHeader(m_data.header) )
                {
                        ++num_frames_fixed;
                }
                if (badheaderworkaround)
                {
                        fixBadFrameHeader(m_data.header);
                }
                if ( (i = DAEEventList::checkEventHeader(&m_data.header, error_message)) != 0 )
                {
                        LOGSTR_WARNING("Event header invalid: " << error_message);
                        DAEEventList::printEventHeader(&m_data.header, m_logstr->warning());
                        m_data.header_words = 0;  // so it will get completely re-read next time
                        n = DAEEventList::findValidEventHeaderOffset(m_buffer + num_words, m_len_buffer - num_words, status);
                        if (n == -1)
                        {
                                LOGSTR_WARNING("Cannot find header in buffer");
                                break;
                        }
                        else
                        {
                                num_words += n;
                                LOGSTR_WARNING("Skipped " << n << " words and found header at offset " << num_words);
                                continue;
                        }
                }
                int frame_diff = m_data.header.frame_number - m_data.last_frame;
                if ( frame_diff != 1 )
                {
                        // we have skipped a frame somewhere? it may go backwards if we begin a new run 
                        LOGSTR_WARNING("skipped frame? found " << m_data.header.frame_number << " expected " << m_data.last_frame + 1);
                }
                n = std::min(m_data.header.num_events - m_data.num_events, m_len_buffer - num_words);
                if (n >= 0)
                {
                        for(i=0; i<n; ++i)
                        {
                                if (m_buffer[num_words+i] == DAEEventHeaderFixedMarker)
                                {
                                        LOGSTR_WARNING("found dae marker at wrong place");
                                }
                        }
                        DetectorEvent32* ev = reinterpret_cast<DetectorEvent32*>(m_buffer + num_words);
                        m_all_event_callbacks(&m_data.header, ev, n, m_dc->eventSourceID(), m_dc->getDAE2SpecMap());
                        m_data.raw_events += n;
                        if (m_data.header.info.bad_frame == 0)   
                        {
                                m_good_event_callbacks(&m_data.header, ev, n, m_dc->eventSourceID(), m_dc->getDAE2SpecMap());
                                m_data.good_events += n;
                        }
                        num_words += n;
                        m_data.num_events += n;
                }
                if (m_data.num_events == m_data.header.num_events)   // is frame complete?
                {
                        if (!DAEEventList::isEndRunHeader(&(m_data.header), status))
                        {
                                ++(m_data.raw_frames);
                                m_all_frame_callbacks(&m_data.header, m_dc->eventSourceID());
                                if (m_data.header.info.bad_frame == 0)
                                {
                                        ++(m_data.good_frames);
                                        m_good_frame_callbacks(&m_data.header, m_dc->eventSourceID());
                                }
                                m_data.last_frame = m_data.header.frame_number;
                        }
                        else
                        {
                                m_data.last_frame = m_data.header.frame_number - 1;
                        }
                        m_data.header_words = 0;
//                      LOGSTR_TRACE("frame " << m_last_frame << " complete");
                }
        }
        if ( !m_immediate_stop && (num_words != m_len_buffer) )
        {
                LOGSTR_WARNING("buffer not fully used - remaining words = " << (m_len_buffer - num_words) );
                //throw std::runtime_error("buffer not fully used");
        }
        if ( num_frames_fixed > 0 )
        {
                LOGSTR_INFORMATION("Frame number in event header was fixed " << num_frames_fixed << " times");
        }
}

EventStoreCard::~EventStoreCard()
{
        LOGSTR_DEBUG("Destructor called");
        closeFiles();
}

void EventStoreCard::closeFiles()
{
        LOGSTR_DEBUG("closeFiles called");
        if (m_file.isOpen())
        {
                m_file.close();
//              m_run_number = -1;
//              m_file_name = "";
//              m_info_file_name = "";
        }
}

boost::signals2::connection EventStoreCard::addGoodEventCallback(const EventCallbackSlotType& slot)
{
        LOGSTR_DEBUG("Adding GOOD event callback");
        return m_good_event_callbacks.connect(slot);
}

boost::signals2::connection EventStoreCard::addAllEventCallback(const EventCallbackSlotType& slot) 
{
        LOGSTR_DEBUG("Adding ALL event callback");
        return m_good_event_callbacks.connect(slot);
}

boost::signals2::connection EventStoreCard::addGoodFrameCallback(const FrameCallbackSlotType& slot) 
{
        LOGSTR_DEBUG("Adding GOOD frame callback");
        return m_good_frame_callbacks.connect(slot);
}

boost::signals2::connection EventStoreCard::addAllFrameCallback(const FrameCallbackSlotType& slot) 
{
        LOGSTR_DEBUG("Adding ALL frame callback");
        return m_all_frame_callbacks.connect(slot);
}

boost::signals2::connection EventStoreCard::addNoFrameCallback(const NoFrameCallbackSlotType& slot) 
{
        LOGSTR_DEBUG("Adding NO frame callback");
        return m_no_frame_callbacks.connect(slot);
}


FileEventSource::FileEventSource(int run_number, int source_id) : FileEventSourceBase(), m_source_id(source_id), m_file_words(0)
{
        setLoggerName("FileEventSource");
        m_file_name = EventStoreCard::fileNameForDC(run_number, source_id);
        m_info_file_name = EventStoreCard::infoFileNameForDC(run_number, source_id);
        m_file.open(m_file_name.c_str(), IFile::ReadOnly, _SH_DENYNO);
        if (!m_file.isOpen())
        {
                throw std::runtime_error("fopen");
        }
        EventStoreCard::getInfoFileData(m_info_file_name, m_info);
}

FileEventSource::~FileEventSource()
{
        if (m_file.isOpen())
        {
                m_file.close();
        }
}

uint32_t FileEventSource::getNewEvents(isisU32_t* buffer, uint32_t maxlen, uint64_t& num_unread, bool& run_ended, DAEstatus& status)
{
        uint64_t new_words = nNewEventWords(status);
        uint32_t num_to_read = 0;
        size_t nread = 0;
        run_ended = false;
        if (new_words == 0)
        {
//              LOGSTR_DEBUG("No events - assuming end");
                run_ended = true;
        }
        if (new_words > maxlen)
        {
                num_unread = new_words - maxlen;
                num_to_read = maxlen;
        }
        else
        {
                num_unread = 0;
                num_to_read = static_cast<uint32_t>(new_words);   // as maxlen is uint32_t this can never overflow
        }
        if (num_to_read > 0)
        {
                nread = m_file.read(buffer, sizeof(uint32_t), num_to_read);
                if (nread != num_to_read)
                {
                        LOGSTR_ERROR("Only read " << nread << " out of " << num_to_read << " requested words");
                        //throw std::runtime_error("getNewEvents");
                }
                if (nread == static_cast<size_t>(-1))
                {
                        nread = 0;
                }
        }
        m_file_words += num_to_read;
        return nread;
}

uint64_t FileEventSource::nNewEventWords(DAEstatus& status)
{
        return m_info.file_words - m_file_words;
}

void FileEventSource::updateDAEEventLastRead(DAEstatus& status)
{
}

int FileEventSource::eventSourceID() const
{
        return m_source_id;
}

const int* FileEventSource::getDAE2SpecMap()
{
        return NULL;
}

isisU32_t FileEventSource::getLastAddressRead()
{
        return 0;
}

void FileEventSource::setLastAddressRead(isisU32_t addr)
{
}

void FileEventSource::readInfo(EventStore::FileEventSourceInfo& info)
{
        info.raw_frames = m_info.raw_frames;
        info.good_frames = m_info.good_frames;
        info.good_events = m_info.good_events;
        info.raw_events = m_info.raw_events;
}

ISISRAWEventSource::ISISRAWEventSource(ISISRAW* iraw, int source_id, time_t run_start) : FileEventSourceBase(), m_file(iraw), 
                m_source_id(source_id), m_events(0), m_events_sent(0), m_pos(0), m_pos_sent(0), m_last_frame(-1)
{
        setLoggerName("ISISRAWEventSource");
        DAEstatus status;
        m_good_ua = m_file->rpb.r_gd_prtn_chrg;
        m_good_frames = m_file->rpb.r_goodfrm;
        m_run_start = run_start;
        int i, j, k, frame;
        int ndet = m_file->i_det;
        int nper = m_file->t_nper;
        int nsp1 = m_file->t_nsp1;
        int ntc1 = m_file->t_ntc1;
        isisU32_t* dat1 = m_file->dat1;
        int persize = (nsp1 + 1) * (ntc1 + 1);
        int maxcrate = 0;
        for(i=0; i < ndet; i++)
        {
                maxcrate = std::max(maxcrate, m_file->crat[i]);
        }
        int maxspec = nsp1 + 1;
        int* last_card_spectrum = new int[maxcrate+1];
        memset(last_card_spectrum, 0, (maxcrate+1) * sizeof(int));
        m_spec_lookup = new int[maxspec];
        memset(m_spec_lookup, 0, maxspec * sizeof(int));
        m_card_lookup = new int[maxspec];
        memset(m_card_lookup, 0, maxspec * sizeof(int));
        int card, spec, offset;
        // we need to pick only one card for all spectrum 0 events
        m_card_lookup[0] = maxcrate;
        for(i=0; i < ndet; i++)
        {
                card = m_file->crat[i];
                spec = m_file->spec[i];
                if (spec > 0)
                {
                        m_card_lookup[spec] = card;
                }
        }
        // dae2 card spectra are assigned in ascending order to dae1 spectra on the card
        m_spec_lookup[0] = 0;
        for(i=1; i < maxspec; ++i)
        {
                card = m_card_lookup[i];
                m_spec_lookup[i] = ++(last_card_spectrum[card]);
        }
        delete []last_card_spectrum;
        for(i=0; i<nper; ++i)
        {
                for(j=0; j<nsp1+1; ++j)
                {
                        if (m_card_lookup[j] == m_source_id)
                        {
                                for(k=0; k<ntc1+1; ++k)
                                {
                                        offset = k + j*(ntc1+1) + i*persize;
                                        m_events += dat1[offset];
                                }
                        }
                }
        }
        m_frame_events.resize(m_good_frames);
        for(i=1; i <= m_events; ++i)
        {
                ++(m_frame_events[getFrame(i)]);
        }
        LOGSTR_INFORMATION("Found " << m_events << " events on card");
//      LOGSTR_INFORMATION("dae1 spectrum 1 on card " << m_card_lookup[1] << " dae2 spec " << m_spec_lookup[1]);
//      LOGSTR_INFORMATION("dae1 spectrum 20 on card " << m_card_lookup[20] << " dae2 spec " << m_spec_lookup[20]);
}

ISISRAWEventSource::~ISISRAWEventSource()
{
}

uint32_t ISISRAWEventSource::getNewEvents(isisU32_t* buffer, uint32_t maxlen, uint64_t& num_unread, bool& run_ended, DAEstatus& status)
{
        uint64_t new_words = nNewEventWords(status);
        run_ended = false;
        if (new_words == 0)
        {
//              LOGSTR_DEBUG("No events - assuming end");
                run_ended = true;
                return 0;
        }
        else
        {
                return readEvents(buffer, maxlen, num_unread);
        }
}

uint64_t ISISRAWEventSource::nNewEventWords(DAEstatus& status)
{
        return m_events - m_events_sent;
}

void ISISRAWEventSource::updateDAEEventLastRead(DAEstatus& status)
{
}

int ISISRAWEventSource::eventSourceID() const
{
        return m_source_id;
}

const int* ISISRAWEventSource::getDAE2SpecMap()
{
        return NULL;
}

isisU32_t ISISRAWEventSource::getLastAddressRead()
{
        return 0;
}

void ISISRAWEventSource::setLastAddressRead(isisU32_t addr)
{
}

void ISISRAWEventSource::readInfo(EventStore::FileEventSourceInfo& info)
{
        info.raw_frames = m_good_frames; 
        info.good_frames = m_good_frames; 
        info.good_events = m_events;
        info.raw_events = m_events;   
}


int ISISRAWEventSource::readEvents(isisU32_t* buffer, uint32_t max_num_to_read, uint64_t& num_unread)
{
        DetectorEvent32 ev;
        int num_sent = 0;
        int spec;
        int ntc1 = m_file->t_ntc1;
        int current_frame;
        while(num_sent < max_num_to_read)
        {
                if ( m_pos >= (m_file->t_nsp1+1) * (m_file->t_ntc1+1) )
                {
                        break;
                }
                spec = m_pos / (ntc1 + 1);
                if ( (m_card_lookup[spec] != m_source_id) || ((m_file->dat1[m_pos] - m_pos_sent) == 0) )
                {
                        ++m_pos;
                        m_pos_sent = 0;
                        continue;
                }
                current_frame = getFrame(m_events_sent+1);  // frame for next event, hence +1
                if (current_frame > m_last_frame)
                {
                        if ( (max_num_to_read - num_sent) < DAE_EVENT_HEADER_WORDS )
                        {
                                break;
                        }
                        sendHeader(buffer+num_sent, current_frame);
                        num_sent += DAE_EVENT_HEADER_WORDS;
                        m_last_frame = current_frame;
                        continue;   // we need to go back round and check we have space for events after header
                }
                ev.spectrum = m_spec_lookup[spec];
                ev.time_channel = m_pos % (ntc1 + 1);
                memcpy(buffer+num_sent, reinterpret_cast<isisU32_t*>(&ev), sizeof(ev)); 
                ++m_pos_sent;
                ++m_events_sent;
                ++num_sent;
        }
        num_unread = m_events - m_events_sent;
        return num_sent;
}

#define PPP_TO_UAMPH    1.738E-6

int ISISRAWEventSource::sendHeader(isisU32_t* buffer, int current_frame)
{
        DAEEventHeader header;
        const int isis_frequency = 10;  // in Hz - 10 means TS2, 50 means TS1
        memset(&header, 0 , sizeof(header));
        header.frame_number = current_frame;
        header.marker0 = static_cast<uint32_t>(DAEEventHeaderFixedMarker);
        header.marker1 = static_cast<uint32_t>(AllValuesFilledInMarker);
        header.info.header_length = DAE_EVENT_HEADER_WORDS;
        header.period = 0;
        header.num_events = m_frame_events[current_frame];
        header.protons = m_good_ua / ((float)m_good_frames * PPP_TO_UAMPH);
        DAEEventList::OffsetToDAETime(m_run_start, (double)current_frame / (double)isis_frequency, header.time);
        header.info.header_type = 0;
        memcpy(buffer, reinterpret_cast<isisU32_t*>(&header), DAE_EVENT_HEADER_WORDS * 4); 
        return 0;
}

int ISISRAWEventSource::getFrame(uint32_t event_count)
{
        int current_frame = floor((double)m_good_frames * (double)(event_count-1) / (double)m_events);
        if (current_frame >= m_good_frames)
        {
                current_frame = m_good_frames - 1;
        }
        return current_frame;
}