event_store.h

Go to the documentation of this file.

#ifndef EVENT_STORE_H
#define EVENT_STORE_H

#include "IsisBase.h"
#include "icp_interfaces.h"
#include "isisvme.h"
#include "dae_events.h"
#include "SingleFile.h"
#include "SplitFile.h"

class EventStoreCard;
class FileEventSource;

typedef boost::signals2::signal< void(const DAEEventHeader*, const DetectorEvent32*, int, int, const int*) > EventCallbackType; 
typedef EventCallbackType::slot_type EventCallbackSlotType; 

typedef boost::signals2::signal< void(const DAEEventHeader*, int) > FrameCallbackType; 
typedef FrameCallbackType::slot_type FrameCallbackSlotType; 

enum NoFrameCallbackMode { NoFrameRunning = 0, NoFrameStop = 1 };
typedef boost::signals2::signal< void(bool, NoFrameCallbackMode, int) > NoFrameCallbackType; 
typedef NoFrameCallbackType::slot_type NoFrameCallbackSlotType; 

class FileEventSourceBase;
class ISISRAW;

class EventStore : public ISIS::Base
{
private:
        static const int m_mempool_size;
        Poco::MemoryPool m_mempool;
        Poco::ThreadPool m_threads;
        std::vector<FileEventSourceBase*> m_sources;   // list of ones we created so we can remove. Should really use shared pointers...
        struct EventStoreCardHolder
        {
                EventStoreCard* card;
                bool enable;
                EventStoreCardHolder() : card(NULL), enable(false) { }
                EventStoreCardHolder(EventStoreCard* card_, bool enable_) : card(card_), enable(enable_) { }
                // card freed explicitly in destructore of EventStore
        };
        std::vector<EventStoreCardHolder> m_stores;
        static Poco::Path m_base_dir;
        std::list<boost::signals2::connection> m_callbacks;

        template<typename SLOT>
        std::list<boost::signals2::connection> addCallback(boost::signals2::connection (EventStoreCard::*func)(const SLOT&), const SLOT& slot);

        void clearAllCallbacks() { std::for_each(m_callbacks.begin(), m_callbacks.end(), boost::bind(&boost::signals2::connection::disconnect, _1)); m_callbacks.clear(); } 
        void clearSources();
        void clearStores();
        void findUnusedFileInputSources(int run_number, std::set<std::string>& files);
        void invokeNoFrameCallbacks(NoFrameCallbackMode mode);

public:
        struct FileEventSourceInfo
        {
                uint64_t raw_events;
                uint64_t good_events;
                uint32_t raw_frames;
                uint32_t good_frames;
                FileEventSourceInfo()
                { 
                        zero(); 
                }
                void zero() 
                {
                        raw_events = good_events = raw_frames = good_frames = 0; 
                }
                void add(const FileEventSourceInfo& f) 
                {
                        
                        raw_events += f.raw_events;
                        good_events += f.good_events;
                        raw_frames = std::max(f.raw_frames, raw_frames);
                        good_frames = std::max(f.good_frames, good_frames);
                }
        };
        EventStore();
        isisU32_t* getBuffer();
        void releaseBuffer(isisU32_t* buffer);
        uint32_t bufferSize() { return m_mempool_size / sizeof(isisU32_t); }
        ~EventStore();
        void addCards(const std::vector<DAEEventSource*>& det_list);
        void addCard(DAEEventSource* det_card, bool enable = false);
        std::list<boost::signals2::connection> addGoodEventCallback(const EventCallbackSlotType& slot);  
        std::list<boost::signals2::connection> addAllEventCallback(const EventCallbackSlotType& slot); 
        std::list<boost::signals2::connection> addGoodFrameCallback(const FrameCallbackSlotType& slot);  
        std::list<boost::signals2::connection> addAllFrameCallback(const FrameCallbackSlotType& slot); 
        std::list<boost::signals2::connection> addNoFrameCallback(const NoFrameCallbackSlotType& slot); 
        void clearCallbacks(const std::list<boost::signals2::connection>& callback_list)
        {
                std::for_each(callback_list.begin(), callback_list.end(), boost::bind(&boost::signals2::connection::disconnect, _1)); 
        }
        void start();
        void stop(bool immediate);
        bool endHeaderSeen();
        uint64_t nNewEventWords(DAEstatus& status);
        void clear();
        void setOutputFileName(int run_number, bool new_run, DAEstatus& status);
        void addFileInputSource(int run_number, FileEventSourceInfo& info);
        void addRawFileInputSource(ISISRAW* iraw, int source_id, time_t run_start);
        void setCardStatus(int id, bool enable);
        void setCardStatus(bool enable);
        uint64_t numRawEvents();
        uint64_t numGoodEvents();
        void getEventSourceIDs(std::vector<int>& ids);
        void getEnabledEventSourceIDs(std::vector<int>& ids);
        static const Poco::Path& baseDir()
        {
                if (m_base_dir.toString().size() == 0)
                {
                        m_base_dir = Poco::Util::Application::instance().config().getString("isisicp.eventdir", "c:\\data\\events");
                }
                return m_base_dir; 
        }

        static Poco::Path baseRunDir(int run_number)
        {
                Poco::Path the_dir(baseDir(), "");
                the_dir.pushDirectory(Poco::format("run_%d", run_number));
                return the_dir;
        }

        static void deleteOutputFile(int run_number, DAEstatus& status);
        void closeOutputFile(DAEstatus& status);
        void closeAndDeleteOutputFile(int run_number, DAEstatus& status);
        static int maxNumCallbackEvents() { return m_mempool_size / sizeof(DetectorEvent32); }     // maximum number of events in a callback

private:
        void clearOutputFile(DAEstatus& status);
        static void findFileInputSources(int run_number, std::set<std::string>& files);
        static int eventSourceIDFromFileInputSource(const std::string& file);

};


struct InfoFileData
{
        uint64_t file_words;                    
        uint64_t good_events;                   
        uint64_t raw_events;                    
        DAEEventHeader header;                  
        uint32_t last_frame;                    
        uint32_t num_events;                    
        uint32_t header_words;                  
        uint32_t good_frames;                   
        uint32_t raw_frames;                    
        uint32_t last_address_read;             
        bool end_header_present;
        void newFile()
        {
                memset(this, 0, sizeof(InfoFileData));
                end_header_present = false;
                --last_frame;
        }
};

class EventStoreCard : public ISIS::Base, public Poco::Runnable
{
public:
        boost::signals2::connection addGoodEventCallback(const EventCallbackSlotType& slot);  
        boost::signals2::connection addAllEventCallback(const EventCallbackSlotType& slot);  

        boost::signals2::connection addGoodFrameCallback(const FrameCallbackSlotType& slot);  
        boost::signals2::connection addAllFrameCallback(const FrameCallbackSlotType& slot);  

        boost::signals2::connection addNoFrameCallback(const NoFrameCallbackSlotType& slot);  

        void flushEvents(DAEstatus& status);

        void checkAndHistogramEvents();

        void run();

        void stop(bool immediate) { m_keep_going = false; m_immediate_stop = immediate; }

        EventStoreCard(EventStore* es, DAEEventSource* dc);

        bool endHeaderSeen() const { return m_data.end_header_present; }
        void setOutputFileName(int run_number, bool new_run, DAEstatus& status);
        void clearOutputFile(DAEstatus& status);
        int eventSourceID() const { return (m_dc != NULL ? m_dc->eventSourceID() : -1); }

        std::string getFileName() { return m_file_name; }
        static std::string fileNameForDC(int run_number, int dc_number);
        static std::string infoFileNameForDC(int run_number, int dc_number);
        static void getInfoFileData(const std::string& file_name, InfoFileData& info);
        void closeFiles();
        uint64_t numRawEvents() { return m_data.raw_events; }
        uint64_t numGoodEvents() { return m_data.good_events; }
        uint64_t nNewEventWords(DAEstatus& status);
        void invokeNoFrameCallbacks(NoFrameCallbackMode mode);
        ~EventStoreCard();

private:
        InfoFileData m_data;
        bool m_keep_going;
        bool m_immediate_stop;
//      bool m_end_header_present;
//      unsigned m_num_events; ///< total number of events read in frame m_header
//      DAEEventHeader m_header;    ///< header of current frame being processed
//      unsigned m_header_words;    ///< number of words of header read, if < DAE_EVENT_HEADER_WORDS then we have a partial header
//      unsigned m_last_frame;
//      uint32_t m_file_words;
        int m_run_number;
        Poco::ThreadPool m_threads;
        EventStore* m_store;
        DAEEventSource* m_dc;
        isisU32_t* m_buffer;      
        uint32_t m_len_buffer;    
        SplitFile m_file;
        std::string m_file_name;
        std::string m_info_file_name;
        EventCallbackType m_good_event_callbacks; 
        EventCallbackType m_all_event_callbacks; 
        FrameCallbackType m_good_frame_callbacks; 
        FrameCallbackType m_all_frame_callbacks; 
        NoFrameCallbackType m_no_frame_callbacks; 
        void updateFilePosition(DAEstatus& status);
        bool fixFrameHeader(DAEEventHeader& head);
        void fixBadFrameHeader(DAEEventHeader& head);
        virtual std::string loggingPrefix() const { return Poco::format("Source %d: ", eventSourceID()); }
        // things we do not want to do
        EventStoreCard() : m_dc(NULL), m_store(NULL), m_buffer(NULL) { }
        void operator=(const EventStoreCard&) { }
        EventStoreCard(const EventStoreCard&) { }
};

class FileEventSourceBase : public ISIS::Base, public DAEEventSource 
{
public:
        virtual ~FileEventSourceBase() { }
};

class FileEventSource : public FileEventSourceBase
{
        std::string m_file_name;
        std::string m_info_file_name;
        uint64_t m_file_words;
        InfoFileData m_info;
        int m_source_id;
        SplitFile m_file;

public:
        FileEventSource(int run_number, int source_id);
        virtual ~FileEventSource();
        uint32_t getNewEvents(isisU32_t* buffer, uint32_t maxlen, uint64_t& num_unread, bool& run_ended, DAEstatus& status);
        uint64_t nNewEventWords(DAEstatus& status);
        void updateDAEEventLastRead(DAEstatus& status);
    int eventSourceID() const;
        const int* getDAE2SpecMap();
        void setLastAddressRead(isisU32_t addr);
        isisU32_t getLastAddressRead();
        void readInfo(EventStore::FileEventSourceInfo& info);
private:
        std::string loggingPrefix() const { return Poco::format("Source %d: ", m_source_id); }
        // things we do not want to do
        FileEventSource() { }
        void operator=(const FileEventSource&) { }
        FileEventSource(const FileEventSource&) { }
};

class ISISRAW;

class ISISRAWEventSource : public FileEventSourceBase 
{
        uint32_t m_events;  
        int m_source_id;
        ISISRAW* m_file;
        int* m_spec_lookup;
        int* m_card_lookup;
        uint32_t m_events_sent;  
        uint32_t m_pos;   
        uint32_t m_pos_sent;   
        float m_good_ua; 
        int m_good_frames;
        int m_last_frame;
        time_t m_run_start;
        std::vector<int> m_frame_events;

        int readEvents(isisU32_t* buffer, uint32_t max_num_to_read, uint64_t& num_unread);
        int getFrame(uint32_t event_count);
        int sendHeader(isisU32_t* buffer, int current_frame);

public:
        ISISRAWEventSource(ISISRAW* iraw, int source_id, time_t run_start);
        virtual ~ISISRAWEventSource();
        uint32_t getNewEvents(isisU32_t* buffer, uint32_t maxlen, uint64_t& num_unread, bool& run_ended, DAEstatus& status);
        uint64_t nNewEventWords(DAEstatus& status);
        void updateDAEEventLastRead(DAEstatus& status);
    int eventSourceID() const;
        const int* getDAE2SpecMap();
        void setLastAddressRead(isisU32_t addr);
        isisU32_t getLastAddressRead();
        void readInfo(EventStore::FileEventSourceInfo& info);
private:
        std::string loggingPrefix() const { return Poco::format("Source %d: ", m_source_id); }
};

#endif /* EVENT_STORE_H */