/afs/cern.ch/work/a/aaltunda/public/www/CMSSW_6_2_5/src/EventFilter/ResourceBroker/interface/SharedResources.h

Go to the documentation of this file.

//
// SharedResources.h
// -------
//
// Resources shared between FSM states.
//
// Created on: Sep 21, 2011
//                                                                              Andrei Spataru : aspataru@cern.ch

#ifndef RBSHAREDRESOURCES_H_
#define RBSHAREDRESOURCES_H_

#include "EventFilter/ResourceBroker/interface/RBStateMachine.h"
#include "EventFilter/ResourceBroker/interface/BUProxy.h"
#include "EventFilter/ResourceBroker/interface/SMProxy.h"
#include "EventFilter/ResourceBroker/interface/IPCManager.h"
#include "EventFilter/ResourceBroker/interface/FUResource.h"
#include "EventFilter/ResourceBroker/interface/CommandQueue.h"

#include "EventFilter/Utilities/interface/IndependentWebGUI.h"

#include "boost/statechart/event_base.hpp"
#include <boost/shared_ptr.hpp>

#include "xdata/InfoSpace.h"
#include "xdata/Integer32.h"
#include "xdata/UnsignedInteger32.h"
#include "xdata/Integer.h"
#include "xdata/Double.h"
#include "xdata/Boolean.h"
#include "xdata/String.h"

#include "toolbox/task/Action.h"
#include "toolbox/task/WorkLoop.h"
#include "toolbox/task/WorkLoopFactory.h"

#include <semaphore.h>
#include <iostream>
#include <queue>

namespace evf {

class FUResourceBroker;

namespace rb_statemachine {

typedef boost::shared_ptr<boost::statechart::event_base> EventPtr;

// forward declaration of state classes
class Halted;
class Configuring;
class Ready;
class Stopped;
class Enabling;
class Enabled;
class Running;
class Stopping;
class Halting;
class Normal;
class Failed;

class SharedResources: public toolbox::lang::Class {

public:
        SharedResources(Logger log);
        ~SharedResources();

        // in SharedResources because it's called by Configuring and Stopping
        void configureResources(xdaq::Application* app);

        // in SharedResources because it's called by Configuring and Stopping
        void reset();
        void updateGUIExternalState(std::string newState) {
                //lock();
                gui_->updateExternalState(newState);
                //unlock();
        }
        void updateGUIInternalState(std::string newState) {
                //lock();
                gui_->updateInternalState(newState);
                //unlock();
        }
        void setFsmPointer(RBStateMachine* const fsm) {
                fsm_ = fsm;
        }

        void cancelAllWorkloops();

        /*
         * CONCURRENT ACCESS
         */
        void lock() {
                while (0 != sem_wait(&lock_)) {
                        if (errno != EINTR) {
                                LOG4CPLUS_ERROR(log_, "Cannot obtain lock on sem LOCK!");
                        }
                }
        }
        void unlock() {
                sem_post(&lock_);
        }

        void lockRSAccess() {
                while (0 != sem_wait(&accessToResourceStructureLock_)) {
                        if (errno != EINTR) {
                                LOG4CPLUS_ERROR(log_, "Cannot obtain lock on sem accessToResourceStructureLock_!");
                        }
                }
        }

        int tryLockRSAccess() {
                if (sem_trywait(&accessToResourceStructureLock_)) return -1;
                return 0;
        }

        void unlockRSAccess() {
                sem_post(&accessToResourceStructureLock_);
        }

        void printWorkLoopStatus();

private:

        /*
         * Workloop starters and dispatch actions.
         * Used by friend state classes.
         */
        void startMonitoringWorkLoop() throw (evf::Exception);
        bool monitoring(toolbox::task::WorkLoop* wl);
        void startWatchingWorkLoop() throw (evf::Exception);
        bool watching(toolbox::task::WorkLoop* wl);
        double deltaT(const struct timeval *start, const struct timeval *end);

        void startSendDataWorkLoop() throw (evf::Exception);
        bool sendData(toolbox::task::WorkLoop* wl);

        void startSendDqmWorkLoop() throw (evf::Exception);
        bool sendDqm(toolbox::task::WorkLoop* wl);

        void startDiscardWorkLoop() throw (evf::Exception);
        bool discard(toolbox::task::WorkLoop* wl);

        void goToFailedState(evf::Exception& e);

private:

        typedef toolbox::task::WorkLoop WorkLoop_t;
        typedef toolbox::task::ActionSignature ActionSignature_t;

        /*
         * Workloops and action signatures
         */

        // workloop / action signature for monitoring
        WorkLoop_t *wlMonitoring_;
        ActionSignature_t *asMonitoring_;
        // workloop / action signature for watching
        WorkLoop_t *wlWatching_;
        ActionSignature_t *asWatching_;

        WorkLoop_t *wlSendData_;
        ActionSignature_t *asSendData_;
        WorkLoop_t *wlSendDqm_;
        ActionSignature_t *asSendDqm_;
        WorkLoop_t *wlDiscard_;
        ActionSignature_t *asDiscard_;

        //
        // member data
        //

        // pointer to FSM
        RBStateMachine* fsm_;

        // Hyper DAQ Independent web GUI
        IndependentWebGUI* gui_;

        // Command Queue containing state machine events
        CommandQueue commands_;

        // application logger
        Logger log_;

        // BuilderUnit (BU) to receive raw even data from
        BUProxy *bu_;

        // StorageManager (SM) to send selected events to
        SMProxy *sm_;

        // memory pool for bu <-> fu comunication messages
        toolbox::mem::Pool* i2oPool_;

        // managed resources
        IPCManager* ipcManager_;
        IPCMethod* resourceStructure_;

        // application identifier
        std::string sourceId_;

        // monitored parameters
        xdata::UnsignedInteger32 runNumber_;

        xdata::Double deltaT_;
        xdata::UnsignedInteger32 deltaN_;
        xdata::Double deltaSumOfSquares_;
        xdata::UnsignedInteger32 deltaSumOfSizes_;

        xdata::Double throughput_;
        xdata::Double rate_;
        xdata::Double average_;
        xdata::Double rms_;

        // monitored counters
        xdata::UnsignedInteger32 nbAllocatedEvents_;
        xdata::UnsignedInteger32 nbPendingRequests_;
        xdata::UnsignedInteger32 nbReceivedEvents_;
        xdata::UnsignedInteger32 nbProcessedEvents_;
        xdata::UnsignedInteger32 nbSentEvents_;
        xdata::UnsignedInteger32 nbSentDqmEvents_;
        xdata::UnsignedInteger32 nbSentErrorEvents_;
        xdata::UnsignedInteger32 nbPendingSMDiscards_;
        xdata::UnsignedInteger32 nbPendingSMDqmDiscards_;
        xdata::UnsignedInteger32 nbDiscardedEvents_;

        // UPDATED
        xdata::UnsignedInteger32 nbReceivedEol_;
        xdata::UnsignedInteger32 highestEolReceived_;
        xdata::UnsignedInteger32 nbEolPosted_;
        xdata::UnsignedInteger32 nbEolDiscarded_;

        xdata::UnsignedInteger32 nbLostEvents_;
        xdata::UnsignedInteger32 nbDataErrors_;
        xdata::UnsignedInteger32 nbCrcErrors_;
        xdata::UnsignedInteger32 nbTimeoutsWithEvent_;
        xdata::UnsignedInteger32 nbTimeoutsWithoutEvent_;
        xdata::UnsignedInteger32 dataErrorFlag_;

        // standard parameters
        xdata::Boolean segmentationMode_;
        xdata::Boolean useMessageQueueIPC_;
        xdata::UnsignedInteger32 nbClients_;
        xdata::String clientPrcIds_;
        xdata::UnsignedInteger32 nbRawCells_;
        xdata::UnsignedInteger32 nbRecoCells_;
        xdata::UnsignedInteger32 nbDqmCells_;
        xdata::UnsignedInteger32 rawCellSize_;
        xdata::UnsignedInteger32 recoCellSize_;
        xdata::UnsignedInteger32 dqmCellSize_;
        // UPDATE freeResourcesRequiredForAllocate
        xdata::Integer freeResRequiredForAllocate_;

        xdata::Boolean doDropEvents_;
        xdata::Boolean doFedIdCheck_;
        xdata::UnsignedInteger32 doCrcCheck_;
        xdata::UnsignedInteger32 doDumpEvents_;

        xdata::String buClassName_;
        xdata::UnsignedInteger32 buInstance_;
        xdata::String smClassName_;
        xdata::UnsignedInteger32 smInstance_;

        xdata::UnsignedInteger32 resourceStructureTimeout_;
        xdata::UnsignedInteger32 monSleepSec_;
        xdata::UnsignedInteger32 watchSleepSec_;
        xdata::UnsignedInteger32 timeOutSec_;
        xdata::Boolean processKillerEnabled_;
        xdata::Boolean useEvmBoard_;

        xdata::String reasonForFailed_;

        // debug parameters
        xdata::UnsignedInteger32 nbAllocateSent_;
        xdata::UnsignedInteger32 nbTakeReceived_;
        xdata::UnsignedInteger32 nbDataDiscardReceived_;
        xdata::UnsignedInteger32 nbDqmDiscardReceived_;

        // helper variables for monitoring
        struct timeval monStartTime_;
        UInt_t nbSentLast_;
        uint64_t sumOfSquaresLast_;
        UInt_t sumOfSizesLast_;

        // lock
        sem_t lock_;
        EvffedFillerRB *frb_;

        bool shmInconsistent_;

        // emergency stop protection against I2O access
        sem_t accessToResourceStructureLock_;
        bool allowI2ODiscards_;

        /*
         * FRIENDS
         */
        friend class evf::FUResourceBroker;

        friend class Halted;
        friend class Configuring;
        friend class Ready;
        friend class Stopped;
        friend class Enabling;
        friend class Enabled;
        friend class Running;
        friend class Stopping;
        friend class Halting;
        friend class Normal;
        friend class Failed;
};

typedef boost::shared_ptr<SharedResources> SharedResourcesPtr_t;

} // end namespace rb_statemachine

} //end namespace evf

#endif /* RBSHAREDRESOURCES_H_ */