stor::Processing Class Reference

#include <StateMachine.h>

Inheritance diagram for stor::Processing:

Public Types
typedef bsc::transition< Stop, DrainingQueues >	DT
typedef bsc::in_state_reaction < EndRun, Processing,&Processing::logEndRunRequest >	EndRunIR
typedef boost::mpl::list< DT, EndRunIR >	reactions

Public Member Functions
void	logEndRunRequest (const EndRun &request)
	Processing (my_context)
virtual	~Processing ()
Public Member Functions inherited from stor::Operations
void	moveToFailedState (xcept::Exception &exception) const
void	noFragmentToProcess () const
	Operations ()
void	processI2OFragment (I2OChain &frag) const
std::string	stateName () const
virtual	~Operations ()=0

Private Member Functions
virtual void	do_entryActionWork ()
virtual void	do_exitActionWork ()
virtual void	do_moveToFailedState (xcept::Exception &exception) const
virtual void	do_noFragmentToProcess () const
virtual void	do_processI2OFragment (I2OChain &frag) const
virtual std::string	do_stateName () const

Additional Inherited Members
Protected Member Functions inherited from stor::Operations
void	safeEntryAction ()
void	safeExitAction ()

Detailed Description

Processing state

Author:

mommsen

Revision:

1.10.12.1

Date:

2011/03/07 11:33:04

Definition at line 503 of file StateMachine.h.

Member Typedef Documentation

typedef bsc::transition<Stop,DrainingQueues> stor::Processing::DT

Definition at line 510 of file StateMachine.h.

typedef bsc::in_state_reaction<EndRun,Processing,&Processing::logEndRunRequest> stor::Processing::EndRunIR

Definition at line 511 of file StateMachine.h.

typedef boost::mpl::list<DT,EndRunIR> stor::Processing::reactions

Definition at line 512 of file StateMachine.h.

Constructor & Destructor Documentation

Processing::Processing

(

my_context

c

)

Definition at line 22 of file Processing.cc.

References stor::Operations::safeEntryAction().

                                    : my_base(c)
{
  safeEntryAction();
}

Processing::~Processing ( )

virtual

Definition at line 35 of file Processing.cc.

References stor::Operations::safeExitAction().

{
  safeExitAction();
}

Member Function Documentation

void Processing::do_entryActionWork ( )

privatevirtual

Implements stor::Operations.

Definition at line 27 of file Processing.cc.

References stor::Operations::stateName().

{
  TransitionRecord tr( stateName(), true );
  outermost_context().updateHistory( tr );
  outermost_context().setExternallyVisibleState( "Enabled" );
  outermost_context().getNotifier()->reportNewState( "Enabled" );
}

void Processing::do_exitActionWork ( )

privatevirtual

Implements stor::Operations.

Definition at line 40 of file Processing.cc.

References stor::Operations::stateName().

{
  TransitionRecord tr( stateName(), false );
  outermost_context().updateHistory( tr );
}

void Processing::do_moveToFailedState ( xcept::Exception &  exception ) const

privatevirtual

Implements stor::Operations.

Definition at line 51 of file Processing.cc.

{
  outermost_context().getSharedResources()->moveToFailedState( exception );
}

void Processing::do_noFragmentToProcess ( ) const

privatevirtual

Reimplemented from stor::Operations.

Definition at line 114 of file Processing.cc.

References stor::WorkerThreadParams::staleFragmentTimeOut_.

{
  I2OChain staleEvent;

  WorkerThreadParams workerParams =
    outermost_context().getSharedResources()->configuration_->getWorkerThreadParams();
  bool gotStaleEvent = 
    outermost_context().getFragmentStore()->
    getStaleEvent(staleEvent, workerParams.staleFragmentTimeOut_);

  if ( gotStaleEvent )
  {
    outermost_context().getSharedResources()->discardManager_->sendDiscardMessage(staleEvent);
    outermost_context().getEventDistributor()->addEventToRelevantQueues(staleEvent);
  }
  outermost_context().getEventDistributor()->checkForStaleConsumers();
}

void Processing::do_processI2OFragment ( I2OChain &  frag ) const

privatevirtual

Reimplemented from stor::Operations.

Definition at line 62 of file Processing.cc.

References stor::FragmentStore::addFragment(), stor::I2OChain::assertRunNumber(), stor::AlarmHandler::ERROR, stor::FragmentStore::memoryUsed(), stor::Operations::noFragmentToProcess(), inputsource_file_cfi::runNumber, stor::ThroughputMonitorCollection::setFragmentStoreMemoryUsed(), stor::ThroughputMonitorCollection::setFragmentStoreSize(), and stor::FragmentStore::size().

{
  static unsigned int noFragmentCount = 0;

  FragmentStore* fragmentStore = outermost_context().getFragmentStore();
  bool completed = fragmentStore->addFragment(frag);
  if ( completed )
  {
    // The run number check has to be done before the event is added to the
    // queues, as for some event types, e.g. error events, the run number
    // match is enforced.
    try
    {
      uint32_t runNumber = outermost_context().getSharedResources()->configuration_->getRunNumber();
      frag.assertRunNumber(runNumber);
    }
    catch(stor::exception::RunNumberMismatch &e)
    {
      // Just raise an alarm, but continue to process the event
      outermost_context().getSharedResources()->statisticsReporter_->
        alarmHandler()->notifySentinel(AlarmHandler::ERROR, e);
    }
    outermost_context().getEventDistributor()->addEventToRelevantQueues(frag);
    outermost_context().getSharedResources()->discardManager_->sendDiscardMessage(frag);
  }
  else
  {
    // Only do the check every 100th fragment
    // TODO: shall we make this number configurable?
    ++noFragmentCount;
    if ( noFragmentCount >= 100 )
    {
      noFragmentCount = 0;
      this->noFragmentToProcess();
    }
  }

  // 12-Aug-2009, KAB - I put the sampling of the fragment store size
  // *after* the code to add fragments to the store and move them from the
  // fragment store to the relevant queues (when needed) so that the baseline
  // number of events in the fragment store is zero.  For example, when
  // disk writing is slow or stopped, the stream queue fills up, and there
  // is backpressure within the SM, the true number of events in the fragment
  // store is zero, and putting the sampling here reflects that.
  ThroughputMonitorCollection& tmc = 
    outermost_context().getSharedResources()->statisticsReporter_->
    getThroughputMonitorCollection();
  tmc.setFragmentStoreSize(fragmentStore->size());
  tmc.setFragmentStoreMemoryUsed(fragmentStore->memoryUsed());
}

string Processing::do_stateName ( ) const

privatevirtual

Implements stor::Operations.

Definition at line 46 of file Processing.cc.

{
  return std::string( "Processing" );
}

void Processing::logEndRunRequest

(

const EndRun &

request

)

Definition at line 56 of file Processing.cc.

{
  outermost_context().unconsumed_event( request );
}