NeXusEventWriter.cpp

Go to the documentation of this file.

#include "stdafx.h"
#include "isiscrpt.h"
#include "isisraw.h"
#include "icputils.h"
#include "selogger.h"
#include "NeXusEventWriter.h"

static void nxErrorFunc(void* arg, char* text)
{
    DAEstatus* pStatus = (DAEstatus*)arg;
        pStatus->addVa(FAC_CRPT, SEV_ERROR, ERRTYPE_OUTOFMEM, "NEXUS ERROR: %s", text);
}


NeXusEventWriter::NeXusEventWriter() : m_nxcallback(NULL), m_file(NULL)
{
        setLoggerName("NeXusEventWriter");
}

void NeXusEventWriter::start(const ISISCRPT_STRUCT* crpt, IXNeXusFile* file, bool create_empty)
{
        stop(true);
        m_file = file;
        start(crpt, create_empty);
}

void NeXusEventWriter::start(int run_number, bool clear_counters)
{
        throw std::runtime_error("not implemented");
}

void NeXusEventWriter::start(const ISISCRPT_STRUCT* crpt, bool create_empty)
{
        LOGSTR_INFORMATION("getting ready to process events for nexus file");
        EventStore::FileEventSourceInfo info;
        m_es.clear();
        m_es.addFileInputSource(crpt->run_number, info);
        std::vector<int> ev_source_ids;
        m_es.getEnabledEventSourceIDs(ev_source_ids);
        m_nxcallback = new NeXusEventCallback(crpt, m_file, ev_source_ids);
        if (create_empty)
        {
                m_nxcallback->createEmptyFileStructure();
        }
        m_file->makeNewGroup("framelog", "NXcollection", false);
        m_es.addAllEventCallback(boost::bind(&NeXusEventCallback::allEventCallback, m_nxcallback, _1, _2, _3, _4, _5));
        m_es.addAllFrameCallback(boost::bind(&NeXusEventCallback::allFrameCallback, m_nxcallback, _1, _2));
        m_es.addNoFrameCallback(boost::bind(&NeXusEventCallback::noFrameCallbackEnd, m_nxcallback, _1, _2, _3));
        m_es.start();
        LOGSTR_INFORMATION("processing events for nexus file");
}

void NeXusEventWriter::wait()
{
        while (!m_es.endHeaderSeen())
        {
                Poco::Thread::sleep(500);
        }
        m_es.stop(false);
        m_nxcallback->finish();
        delete m_nxcallback;
        m_nxcallback = NULL;
        LOGSTR_INFORMATION("finished events for nexus file");
}

void NeXusEventWriter::stop(bool close_file)
{
        m_es.stop(true);
        m_es.clear();
        if (m_nxcallback != NULL)
        {
                delete m_nxcallback;
                m_nxcallback = NULL;
        }
        if ( close_file && (m_file != NULL) )
        {
                m_file->close();
                delete m_file;
                m_file = NULL;
        }
}

ICPEventWriter::ICPEventWriter() : m_dae(NULL), m_crpt(NULL), m_nxcallback(NULL), m_file(NULL), m_status(NULL), m_run_number_digits(0)
{
        setLoggerName("ICPEventWriter");
}

ICPEventWriter::~ICPEventWriter()
{
        stop();
}

// status needs to be persistent as reference kept in nexus file
void ICPEventWriter::setup(const ISISCRPT_STRUCT* crpt, ISISDAE* dae, int run_number_digits, DAEstatus& status)
{
        LOGSTR_INFORMATION("Setup nexus event writer");
        m_crpt = crpt;
        m_run_number_digits = run_number_digits;
        m_dae = dae;
        m_status = &status;
        remove(filename().c_str());    // if we got restarted during a run, this will be corrupt and useless
}

std::string ICPEventWriter::filename()
{
        if (m_crpt != NULL)
        { 
                return Poco::format("c:\\data\\events%s.tmp", padWithZeros(m_crpt->run_number, m_run_number_digits));
        }
        else
        {
                return "";
        }
}


void ICPEventWriter::start()
{
        if ( !Poco::Util::Application::instance().config().getBool("isisicp.incrementaleventnexus", false) )
        {
                return;
        }
        LOGSTR_INFORMATION("Starting nexus event writer for " << filename());
        // start() may get called twice without a stop() if e.g. BEGIN; ABORT; BEGIN
        stop(true);  // deletes file

        m_file = new IXNeXusFile(nxErrorFunc, m_status);
        m_file->open(filename().c_str(), IXNeXusFile::Write, IXNeXusFile::HDF5);

        NXsetcache(64*1024*1024);
//      m_file->enableOptions(IXNeXusFile::WriteDummyOnNULL);
        m_file->enableCompression(100*NX_COMP_LZW + m_crpt->compression_level);
//      dstatus.addInfoVa(FAC_DAE, "compression level = %d", crpt->compression_level);
        m_file->setSpectraChunkSize(m_crpt->compression_block_size);
//      dstatus.addInfoVa(FAC_DAE, "compression block size = %d", crpt->compression_block_size);

        std::vector<int> ev_source_ids;
        m_dae->getEnabledEventSourceIDs(ev_source_ids);
        m_nxcallback = new NeXusEventCallback(m_crpt, m_file, ev_source_ids);
        m_nxcallback->createEmptyFileStructure();
        m_file->makeNewGroup("framelog", "NXcollection", false);
        std::list<boost::signals2::connection> cb;
        m_callbacks.clear();
        cb = m_dae->addAllEventCallback(boost::bind(&NeXusEventCallback::allEventCallback, m_nxcallback, _1, _2, _3, _4, _5));
        m_callbacks.insert(m_callbacks.begin(), cb.begin(), cb.end());
        cb = m_dae->addAllFrameCallback(boost::bind(&NeXusEventCallback::allFrameCallback, m_nxcallback, _1, _2));
        m_callbacks.insert(m_callbacks.begin(), cb.begin(), cb.end());
        cb = m_dae->addNoFrameCallback(boost::bind(&NeXusEventCallback::noFrameCallbackNull, m_nxcallback, _1, _2, _3));
        m_callbacks.insert(m_callbacks.begin(), cb.begin(), cb.end());
}

void ICPEventWriter::stop(bool delete_file)
{
        if ( !Poco::Util::Application::instance().config().getBool("isisicp.incrementaleventnexus", false) )
        {
                return;
        }
        LOGSTR_INFORMATION("Stopping nexus event writer for " << filename());
        if (m_dae != NULL)
        {
                m_dae->clearCallbacks(m_callbacks);
        }
        m_callbacks.clear();
        if (m_nxcallback != NULL)
        {
                if (!delete_file)
                {
                        m_nxcallback->finish();
                }
                delete m_nxcallback;
                m_nxcallback = NULL;
        }
        if (m_file != NULL)
        {
                m_file->close();
                delete m_file;
                m_file = NULL;
        }
        if (delete_file)
        {
                remove(filename().c_str());
        }
}

extern "C" CRITICAL_SECTION nx_critical;

void ICPEventWriter::snapshotOld(const std::string& name)
{
        static std::string app_dir = Poco::Util::Application::instance().config().getString("application.dir");
        static std::string shadow_user = Poco::Util::Application::instance().config().getString("isisicp.shadow.user");
        static std::string shadow_pass = Poco::Util::Application::instance().config().getString("isisicp.shadow.pass");
        static std::string shadow_snapvol = Poco::Util::Application::instance().config().getString("isisicp.shadow.snapvolume");
        static std::string shadow_copydir = Poco::Util::Application::instance().config().getString("isisicp.shadow.copydir");
        static std::string shadow_prog_name = Poco::Util::Application::instance().config().getString("isisicp.shadow.progname");
        static std::string shadow_dir = Poco::Util::Application::instance().config().getString("isisicp.shadow.dir");
        static std::string comspec = getenv("ComSpec");
        static std::string tmpdir = getenv("TEMP");
        static std::string script_name = Poco::Path(tmpdir, "isisicp_shadow.cmd").toString(Poco::Path::PATH_WINDOWS);
        if ( m_file == NULL || !Poco::File(filename()).exists() )
        {
                LOGSTR_WARNING("Nothing to shadow copy");
                return;
        }
        if ( !Poco::File(shadow_dir).exists() )
        {
                LOGSTR_ERROR("Shadow Directory \"" << shadow_dir << "\" does not exist");
                return;
        }
        // program resides in directory above isisicp e.g. in service/x64 not service/x64/{Debug,Release}
        Poco::Path shadow_prog_path(app_dir, shadow_prog_name);
        shadow_prog_path.pushDirectory("..");
        std::string shadow_prog = shadow_prog_path.toString(Poco::Path::PATH_WINDOWS);
        DWORD len_comp_name = MAX_COMPUTERNAME_LENGTH + 1;
        wchar_t comp_name[MAX_COMPUTERNAME_LENGTH + 1];
        GetComputerNameW(comp_name, &len_comp_name);
        comp_name[len_comp_name] = '\0';
    PROCESS_INFORMATION pi = {0};
    STARTUPINFO         si = {0};
    STARTUPINFOW        siw = {0};
    ZeroMemory( &si, sizeof(si) );
    ZeroMemory( &siw, sizeof(siw) );
    ZeroMemory( &pi, sizeof(pi) );
    si.cb = sizeof(si);
        //si.dwFlags = STARTF_USESHOWWINDOW;
        //si.wShowWindow = SW_MINIMIZE;
    siw.cb = sizeof(siw);
        //siw.dwFlags = STARTF_USESHOWWINDOW;
        //siw.wShowWindow = SW_MINIMIZE;
        BOOL res;
//      std::string command_line_s = Poco::format("\"%s\" -nw -p -script=\"%s\" \"%s\"", shadow_prog, script_name, shadow_snapvol);
        std::string command_line_s = Poco::format("\"%s\" /k \"\"%s\" -tracing -nw -p -script=\"%s\" \"%s\"\"", comspec, 
                shadow_prog, script_name, shadow_snapvol);
        char* command_line = strdup(command_line_s.c_str());
        EnterCriticalSection(&nx_critical);
        m_file->flush();
        if (shadow_user.size() == 0)
        {
                res = CreateProcess(NULL, command_line, NULL, NULL, FALSE, CREATE_NEW_CONSOLE, NULL, NULL, &si, &pi);
        }
        else
        {
                res = CreateProcessWithLogonW(CT2OLE(shadow_user.c_str()), comp_name, CT2OLE(shadow_pass.c_str()), 
                                                                                0, NULL, CT2OLE(command_line), 0, NULL, NULL, &siw, &pi);
        }
        free(command_line);
        if (!res)
        {
                LeaveCriticalSection(&nx_critical); // once shadow created, we can continue writing 
                LOGSTR_ERROR("Shadow error");
                return;
        }
        WaitForSingleObject( pi.hProcess, INFINITE );
        CloseHandle(pi.hProcess);
        CloseHandle(pi.hThread);
        LeaveCriticalSection(&nx_critical); // once shadow created, we can continue writing 
    ZeroMemory( &si, sizeof(si) );
    ZeroMemory( &siw, sizeof(siw) );
    ZeroMemory( &pi, sizeof(pi) );
    si.cb = sizeof(si);
        //si.dwFlags = STARTF_USESHOWWINDOW;
        //si.wShowWindow = SW_MINIMIZE;
    siw.cb = sizeof(siw);
        //siw.dwFlags = STARTF_USESHOWWINDOW;
        //siw.wShowWindow = SW_MINIMIZE;
        std::fstream f;
        f.open(script_name, std::ios::app);
        f << "\"" << shadow_prog <<"\" -tracing -el=%SHADOW_ID_1%,\"" << shadow_dir << "\"" << std::endl;
        Poco::Path fn(filename());
        Poco::Path ff(shadow_dir, fn.getFileName());
        ff.pushDirectory(shadow_copydir);
        f << "COPY /B /Y \"" << ff.toString() << "\" \"" << name << "\"" << std::endl;
        f << "\"" << shadow_prog <<"\" -tracing -ds=%SHADOW_ID_1%" << std::endl;
        f.close();
        command_line_s = Poco::format("\"%s\" /k \"%s\"", comspec, script_name);
        command_line = strdup(command_line_s.c_str());
        if (shadow_user.size() == 0)
        {
                res = CreateProcess(NULL, command_line, NULL, NULL, FALSE, CREATE_NEW_CONSOLE, NULL, NULL, &si, &pi);
        }
        else
        {
                res = CreateProcessWithLogonW(CT2OLE(shadow_user.c_str()), comp_name, CT2OLE(shadow_pass.c_str()), 
                                                                                0, NULL, CT2OLE(command_line), 0, NULL, NULL, &siw, &pi);
        }
        free(command_line);
        if (!res)
        {
                LOGSTR_ERROR("Shadow error");
                remove(script_name.c_str());
                return;
        }
        WaitForSingleObject( pi.hProcess, INFINITE );
        CloseHandle(pi.hProcess);
        CloseHandle(pi.hThread);
        remove(script_name.c_str());
}
        
void ICPEventWriter::snapshot(const std::string& name)
{
        static std::string app_dir = Poco::Util::Application::instance().config().getString("application.dir");
        static std::string shadow_user = Poco::Util::Application::instance().config().getString("isisicp.shadow.user");
        static std::string shadow_pass = Poco::Util::Application::instance().config().getString("isisicp.shadow.pass");
        static std::string shadow_prog_name = Poco::Util::Application::instance().config().getString("isisicp.shadow.progname");
        static std::string shadow_dir = Poco::Util::Application::instance().config().getString("isisicp.shadow.dir");
        static std::string comspec = getenv("ComSpec");
        static std::string tmpdir = getenv("TEMP");
        static std::string script_name = Poco::Path(tmpdir, "isisicp_shadow.cmd").toString(Poco::Path::PATH_WINDOWS);
        if ( m_file == NULL || !Poco::File(filename()).exists() )
        {
                LOGSTR_WARNING("Nothing to shadow copy");
                return;
        }
        if ( !Poco::File(shadow_dir).exists() )
        {
                LOGSTR_ERROR("Shadow Directory \"" << shadow_dir << "\" does not exist");
                return;
        }
        // program resides in directory above isisicp e.g. in service/x64 not service/x64/{Debug,Release}
        Poco::Path shadow_prog_path(app_dir, shadow_prog_name);
        shadow_prog_path.pushDirectory("..");
        std::string shadow_prog = shadow_prog_path.toString(Poco::Path::PATH_WINDOWS);
        DWORD len_comp_name = MAX_COMPUTERNAME_LENGTH + 1;
        wchar_t comp_name[MAX_COMPUTERNAME_LENGTH + 1];
        GetComputerNameW(comp_name, &len_comp_name);
        comp_name[len_comp_name] = '\0';
    PROCESS_INFORMATION pi = {0};
    STARTUPINFO         si = {0};
    STARTUPINFOW        siw = {0};
    ZeroMemory( &si, sizeof(si) );
    ZeroMemory( &siw, sizeof(siw) );
    ZeroMemory( &pi, sizeof(pi) );
    si.cb = sizeof(si);
        //si.dwFlags = STARTF_USESHOWWINDOW;
        //si.wShowWindow = SW_MINIMIZE;
    siw.cb = sizeof(siw);
        //siw.dwFlags = STARTF_USESHOWWINDOW;
        //siw.wShowWindow = SW_MINIMIZE;

        BOOL res;
        Poco::Path src_dir(filename());
        std::string filename_base(src_dir.getFileName());
        src_dir.setFileName("");
        Poco::Path dest_filename(shadow_dir, filename_base);

        std::fstream f;
        f.open(script_name, std::ios::out | std::ios::trunc);
        f << "REM \"" << shadow_prog << "\" /full \"" << src_dir.toString() << "\" \"" << shadow_dir << "\" \"" << filename_base << "\"" << std::endl;
//      f << "COPY /B /Y \"" << ff.toString() << "\" \"" << name << "\"" << std::endl;
        f << "REM REN \"" << dest_filename.toString() << "\" \"" << name << "\"" << std::endl;
        f.close();
        std::string command_line_s = Poco::format("\"%s\" /k \"%s\"", comspec, script_name);
        char* command_line = strdup(command_line_s.c_str());
        EnterCriticalSection(&nx_critical);
        m_file->flush();
        if (shadow_user.size() == 0)
        {
                res = CreateProcess(NULL, command_line, NULL, NULL, FALSE, CREATE_NEW_CONSOLE, NULL, NULL, &si, &pi);
        }
        else
        {
                res = CreateProcessWithLogonW(CT2OLE(shadow_user.c_str()), comp_name, CT2OLE(shadow_pass.c_str()), 
                                                                                0, NULL, CT2OLE(command_line), 0, NULL, NULL, &siw, &pi);
        }
        free(command_line);
        if (!res)
        {
                LeaveCriticalSection(&nx_critical); // once shadow created, we can continue writing 
                LOGSTR_ERROR("Shadow error");
                remove(script_name.c_str());
                return;
        }
        WaitForSingleObject( pi.hProcess, INFINITE );
        CloseHandle(pi.hProcess);
        CloseHandle(pi.hThread);
        LeaveCriticalSection(&nx_critical); // once shadow created, we can continue writing 
        remove(script_name.c_str());
}