source: trunk/FACT++/src/EventBuilderWrapper.h@ 10944

#ifndef FACT_EventBuilderWrapper
#define FACT_EventBuilderWrapper

/*
#if BOOST_VERSION < 104400
#if (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ > 4))
#undef BOOST_HAS_RVALUE_REFS
#endif
#endif
#include <boost/thread.hpp>

using namespace std;
*/

#include <boost/date_time/posix_time/posix_time_types.hpp>

#include "EventBuilder.h"

extern "C" {
    extern void StartEvtBuild();
    extern int CloseRunFile(uint32_t runId, uint32_t closeTime);
}

class EventBuilderWrapper
{
public:
    // FIXME
    static EventBuilderWrapper *This;

private:
    boost::thread fThread;

    enum CommandStates_t // g_runStat
    {
        kAbort      = -2,  // quit as soon as possible ('abort')
        kExit       = -1,  // stop reading, quit when buffered events done ('exit')
        kInitialize =  0,  // 'initialize' (e.g. dim not yet started)
        kHybernate  =  1,  // do nothing for long time ('hybernate') [wakeup within ~1sec]
        kSleep      =  2,  // do nothing ('sleep')                   [wakeup within ~10msec]
        kModeFlush  = 10,  // read data from camera, but skip them ('flush')
        kModeTest   = 20,  // read data and process them, but do not write to disk ('test')
        kModeFlag   = 30,  // read data, process and write all to disk ('flag')
        kModeRun    = 40,  // read data, process and write selected to disk ('run')
    };

    MessageImp &fMsg;

    enum
    {
        kCurrent = 0,
        kTotal   = 1
    };

    bool fFitsFormat;

    uint32_t fMaxRun;
    uint32_t fNumEvts[2];

    DimDescribedService fDimFiles;
    DimDescribedService fDimRuns;
    DimDescribedService fDimEvents;
    DimDescribedService fDimCurrentEvent;

public:
    EventBuilderWrapper(MessageImp &msg) : fMsg(msg), 
        fFitsFormat(false), fMaxRun(0),
        fDimFiles ("FAD_CONTROL/FILES",         "X:1", ""),
        fDimRuns  ("FAD_CONTROL/RUNS",          "I:1", ""),
        fDimEvents("FAD_CONTROL/EVENTS",        "I:2", ""),
        fDimCurrentEvent("FAD_CONTROL/CURRENT_EVENT", "I:1", "")
    {
        if (This)
            throw logic_error("EventBuilderWrapper cannot be instantiated twice.");

        This = this;

        memset(fNumEvts, 0, sizeof(fNumEvts));

        Update(fDimRuns,         uint32_t(0));
        Update(fDimCurrentEvent, uint32_t(0));
        Update(fDimEvents,       fNumEvts);
    }
    ~EventBuilderWrapper()
    {
        Abort();
        // FIXME: Used timed_join and abort afterwards
        //        What's the maximum time the eb need to abort?
        fThread.join();
        //fMsg.Info("EventBuilder stopped.");

        for (vector<DataFileImp*>::iterator it=fFiles.begin(); it!=fFiles.end(); it++)
            delete *it;
    }

    void Update(ostringstream &msg, int severity)
    {
        fMsg.Update(msg, severity);
    }

    bool IsThreadRunning()
    {
        return !fThread.timed_join(boost::posix_time::microseconds(0));
    }

    void SetMaxMemory(unsigned int mb) const
    {
        if (mb*1000000<GetUsedMemory())
        {
            // fMsg.Warn("...");
            return;
        }

        g_maxMem = mb*1000000;
    }

    void Start(const vector<tcp::endpoint> &addr)
    {
        if (IsThreadRunning())
        {
            fMsg.Warn("Start - EventBuilder still running");
            return;
        }

        int cnt = 0;
        for (size_t i=0; i<40; i++)
        {
            if (addr[i]==tcp::endpoint())
            {
                g_port[i].sockDef = -1;
                continue;
            }

            // -1:  if entry shell not be used
            //  0:  event builder will connect but ignore the events
            //  1:  event builder will connect and build events
            g_port[i].sockDef                  = 1;

            g_port[i].sockAddr.sin_family      = AF_INET;
            g_port[i].sockAddr.sin_addr.s_addr = htonl(addr[i].address().to_v4().to_ulong());
            g_port[i].sockAddr.sin_port        = htons(addr[i].port());

            cnt++;
        }

//        g_maxBoards = cnt;
        g_actBoards = cnt;

        g_runStat   = kModeRun;

        fMsg.Message("Starting EventBuilder thread");

        fThread = boost::thread(StartEvtBuild);
    }
    void Abort()
    {
        fMsg.Message("Signal abort to EventBuilder thread...");
        g_runStat = kAbort;
    }

    void Exit()
    {
        fMsg.Message("Signal exit to EventBuilder thread...");
        g_runStat = kExit;
    }

    /*
    void Wait()
    {
        fThread.join();
        fMsg.Message("EventBuilder stopped.");
    }*/

    void Hybernate() const { g_runStat = kHybernate; }
    void Sleep()     const { g_runStat = kSleep;     }
    void FlushMode() const { g_runStat = kModeFlush; }
    void TestMode()  const { g_runStat = kModeTest;  }
    void FlagMode()  const { g_runStat = kModeFlag;  }
    void RunMode()   const { g_runStat = kModeRun;   }

    // FIXME: To be removed
    void SetMode(int mode) const { g_runStat = mode; }

    bool IsConnected(int i) const     { return gi_NumConnect[i]==7; }
    bool IsDisconnected(int i) const  { return gi_NumConnect[i]<=0; }
    int  GetNumConnected(int i) const { return gi_NumConnect[i]; }

    size_t GetUsedMemory() const { return gi_usedMem; }

    virtual int CloseOpenFiles() { CloseRunFile(0, 0); return 0; }


    /*
     struct OpenFileToDim
     {
        int code;
        char fileName[FILENAME_MAX];
     };

     SignalRunOpened(runid, filename);
     // Send num open files
     // Send runid, (more info about the run?), filename via dim

     SignalEvtWritten(runid);
     // Send num events written of newest file

     SignalRunClose(runid);
     // Send new num open files
     // Send empty file-name if no file is open

     */

    // -------------- Mapped event builder callbacks ------------------

    class DataFileImp
    {
        uint32_t fRunId;

    public:
        DataFileImp(uint32_t id) : fRunId(id) { }

        virtual bool Write() = 0;
        virtual bool Close() = 0;

        uint32_t GetRunId() const { return fRunId; }
    };


    class DataFileRaw : public DataFileImp
    {
    public:
        DataFileRaw(uint32_t id) : DataFileImp(id)  { }
        ~DataFileRaw() { Close(); }

        virtual bool Write() { return true; }
        virtual bool Close() { return true; }
    };

    class DataFileFits : public DataFileImp
    {
    public:
        DataFileFits(uint32_t id) :DataFileImp(id) { }
        ~DataFileFits() { Close(); }

        virtual bool Write() { return true; }
        virtual bool Close() { return true; }
    };

    vector<DataFileImp*> fFiles;

    template<class T>
        void Update(DimDescribedService &svc, const T &data) const
    {
        cout << "Update: " << svc.getName() << " (" << sizeof(T) << ")" << endl;
        svc.setData(const_cast<T*>(&data), sizeof(T));
        svc.updateService();
    }

    FileHandle_t runOpen(uint32_t runid, RUN_HEAD *h, size_t)
    {
        // Check if file already exists...
        DataFileImp *file = NULL;
        try
        {
            file = fFitsFormat ?
                static_cast<DataFileImp*>(new DataFileFits(runid)) :
                static_cast<DataFileImp*>(new DataFileRaw(runid));
        }
        catch (const exception &e)
        {
            return 0;
        }

        cout << "OPEN_FILE #" << runid << " (" << file << ")" <<  endl;
        cout << " Ver= " << h->Version << endl;
        cout << " Typ= " << h->RunType << endl;
        cout << " Nb = " << h->NBoard << endl;
        cout << " Np = " << h->NPix << endl;
        cout << " NTm= " << h->NTm << endl;
        cout << " roi= " << h->Nroi << endl;

        fFiles.push_back(file);

        if (runid>fMaxRun)
        {
            fMaxRun = runid;
            fNumEvts[kCurrent] = 0;

            Update(fDimRuns,         fMaxRun);
            Update(fDimEvents,       fNumEvts);
            Update(fDimCurrentEvent, uint32_t(0));
        }

        Update(fDimFiles, fFiles.size());

//        fDimFiles.setData(fFiles.size());
//        fDimFiles.update();

        return reinterpret_cast<FileHandle_t>(file);
    }

    int runWrite(FileHandle_t handler, EVENT *e, size_t)
    {
        DataFileImp *file = reinterpret_cast<DataFileImp*>(handler);

        cout << "WRITE_EVENT " << file->GetRunId() << endl;

        cout << " Evt=" << e->EventNum << endl;
        cout << " Typ=" << e->TriggerType << endl;
        cout << " roi=" << e->Roi << endl;
        cout << " trg=" << e->SoftTrig << endl;
        cout << " tim=" << e->PCTime << endl;

        if (!file->Write())
            return -1;

        if (file->GetRunId()==fMaxRun)
        {
            Update(fDimCurrentEvent, e->EventNum);
            fNumEvts[kCurrent]++;
        }

        fNumEvts[kTotal]++;
        Update(fDimEvents, fNumEvts);

        // ===> SignalEvtWritten(runid);
        // Send num events written of newest file

        /* close run runId (all all runs if runId=0) */
        /* return: 0=close scheduled / >0 already closed / <0 does not exist */
        //CloseRunFile(file->GetRunId(), time(NULL)+2) ;

        return 0;
    }

    int runClose(FileHandle_t handler, RUN_TAIL *, size_t)
    {
        DataFileImp *file = reinterpret_cast<DataFileImp*>(handler);

         const vector<DataFileImp*>::iterator it = find(fFiles.begin(), fFiles.end(), file);
        if (it==fFiles.end())
        {
            ostringstream str;
            str << "File handler (" << handler << ") requested to close by event builder doesn't exist.";
            fMsg.Fatal(str);
            return -1;
        }

        ostringstream str;
        str << "CLOSE_RUN requested for " << file->GetRunId() << " (" << file << ")" <<endl;
        fMsg.Debug(str);

        fFiles.erase(it);

        Update(fDimFiles, fFiles.size());

        //fDimFiles.setData(fFiles.size());
        //fDimFiles.update();

        const bool rc = file->Close();
        if (!rc)
        {
            // Error message
        }

        delete file;

        // ==> SignalRunClose(runid);
        // Send new num open files
        // Send empty file-name if no file is open

        return rc ? 0 : -1;
    }
};

EventBuilderWrapper *EventBuilderWrapper::This = 0;

// ----------- Event builder callbacks implementation ---------------
extern "C"
{
    FileHandle_t runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len)
    {
        return EventBuilderWrapper::This->runOpen(irun, runhd, len);
    }

    int runWrite(FileHandle_t fileId, EVENT *event, size_t len)
    {
        return EventBuilderWrapper::This->runWrite(fileId, event, len);
    }

    int runClose(FileHandle_t fileId, RUN_TAIL *runth, size_t len)
    {
        return EventBuilderWrapper::This->runClose(fileId, runth, len);
    }

    void factOut(int severity, int err, char *message)
    {
        //replace(message, message+strlen(message), '\n', ' ');

        // FIXME: Make the output to the console stream thread-safe
        ostringstream str;
        str << "EventBuilder(";
        if (err<0)
            str << "---";
        else
            str << err;
        str << "): " << message;
        EventBuilderWrapper::This->Update(str, severity);
    }

    void factStat(int severity, int err, char* message )
    {
        static string last;
        if (message==last)
            return;

        if (err!=-1)
            factOut(severity, err, message);
        else
        {
            ostringstream str("Status: ");
            str << message;
            EventBuilderWrapper::This->Update(str, severity);
        }

        last = message;
    }

    /*
    void message(int severity, const char *msg)
    {
        EventBuilderWrapper::This->Update(msg, severity);
    }*/
}

#endif