---

stor::DQMEventServer Class Reference

#include <EventFilter/StorageManager/interface/DQMEventServer.h>

List of all members.

Public Member Functions
void	addConsumer (boost::shared_ptr< DQMConsumerPipe > consumer)
	Adds the specified consumer to the event server.
void	clearQueue ()
	DQMEventServer (double maximumRate)
	DQMEventServer constructor.
boost::shared_ptr < DQMConsumerPipe >	getConsumer (uint32 consumerId)
	Returns a shared pointer to the consumer pipe with the specified ID or an empty pointer if the ID was not found.
std::map< uint32, boost::shared_ptr < DQMConsumerPipe > >	getConsumerTable ()
boost::shared_ptr< std::vector < char > >	getDQMEvent (uint32 consumerId)
	Returns the next event for the specified consumer.
void	processDQMEvent (const DQMEventMsgView &eventView)
	Processes the specified event.
	~DQMEventServer ()
	DQMEventServer destructor.
Private Attributes
std::map< uint32, boost::shared_ptr < DQMConsumerPipe > >	consumerTable
int	disconnectedConsumerTestCounter_
struct timeval	lastAcceptedEventTime_
double	minTimeBetweenEvents_
Static Private Attributes
static const double	MAX_ACCEPT_INTERVAL = 86400.0
	Initialize the maximum accept interval.

---

Detailed Description

Definition at line 32 of file DQMEventServer.h.

---

Constructor & Destructor Documentation

DQMEventServer::DQMEventServer

(

double

maximumRate

)

DQMEventServer constructor.

One way of throttling DQMevents is supported: specifying a maximimum allowed rate of accepted event. However it is not yet implemented because it should be a maximum rate of accepted updates, since one update can have multiple DQMEvents with the same folderID but different eventAtUpdate from different FUs are about the same time and we want to accept all these to collate them Needs implementation to handle if already collated. Also means we need a buffer that can hold more than one DQMEvent

Definition at line 35 of file DQMEventServer.cc.

References disconnectedConsumerTestCounter_, lastAcceptedEventTime_, MAX_ACCEPT_INTERVAL, and minTimeBetweenEvents_.

{
  // determine the amount of time that we need to wait between accepted
  // events (to ensure that the event server doesn't send "too many" events
  // to consumers).  The maximum rate specified to this constructor is
  // converted to an interval that is used internally, and the interval
  // is required to be somewhat reasonable.
  /* Note that this is not
   * yet implemented because it should be a maximum rate of accepted updates,
   * since one update can have multiple DQMEvents with the same folderID but
   * different eventAtUpdate from different FUs are about the same time and we
   * want to accept all these to collate them
   * Needs implementation to handle if already collated.
   */
  if (maximumRate < (1.0 / MAX_ACCEPT_INTERVAL))
  {
    minTimeBetweenEvents_ = MAX_ACCEPT_INTERVAL;
  }
  else
  {
    minTimeBetweenEvents_ = 1.0 / maximumRate;  // seconds
  }

  // initialize the last accepted event time to construction time
  struct timezone dummyTZ;
  gettimeofday(&lastAcceptedEventTime_, &dummyTZ);

  // initialize counters
  disconnectedConsumerTestCounter_ = 0;
}

DQMEventServer::~DQMEventServer

(

)

DQMEventServer destructor.

Definition at line 69 of file DQMEventServer.cc.

References lat::endl(), and FDEBUG.

{
  FDEBUG(5) << "Executing destructor for DQMevent server " << std::endl;
}

---

Member Function Documentation

void DQMEventServer::addConsumer

(

boost::shared_ptr< DQMConsumerPipe >

consumer

)

Adds the specified consumer to the event server.

Definition at line 77 of file DQMEventServer.cc.

References consumerTable.

{
  uint32 consumerId = consumer->getConsumerId();
  consumerTable[consumerId] = consumer;
}

void DQMEventServer::clearQueue

(

)

Definition at line 245 of file DQMEventServer.cc.

References consumerTable.

{
  std::map< uint32, boost::shared_ptr<DQMConsumerPipe> >::const_iterator consIter;
  for (consIter = consumerTable.begin();
       consIter != consumerTable.end();
       consIter++)
  {
    boost::shared_ptr<DQMConsumerPipe> consPipe = consIter->second;
    consPipe->clearQueue();
  }
}

boost::shared_ptr< DQMConsumerPipe > DQMEventServer::getConsumer

(

uint32

consumerId

)

Returns a shared pointer to the consumer pipe with the specified ID or an empty pointer if the ID was not found.

Definition at line 87 of file DQMEventServer.cc.

References consumerTable.

{
  // initial empty pointer
  boost::shared_ptr<DQMConsumerPipe> consPtr;

  // lookup the consumer
  std::map< uint32, boost::shared_ptr<DQMConsumerPipe> >::const_iterator consIter;
  consIter = consumerTable.find(consumerId);
  if (consIter != consumerTable.end())
  {
    consPtr = consIter->second;
  }

  // return the pointer
  return consPtr;
}

std::map< uint32, boost::shared_ptr<DQMConsumerPipe> > stor::DQMEventServer::getConsumerTable

(

)

[inline]

Definition at line 43 of file DQMEventServer.h.

References consumerTable.

    { return(consumerTable);}

boost::shared_ptr< std::vector< char > > DQMEventServer::getDQMEvent

(

uint32

consumerId

)

Returns the next event for the specified consumer.

Definition at line 227 of file DQMEventServer.cc.

References consumerTable.

{
  // initial empty buffer
  boost::shared_ptr< vector<char> > bufPtr;

  // lookup the consumer
  std::map< uint32, boost::shared_ptr<DQMConsumerPipe> >::const_iterator consIter;
  consIter = consumerTable.find(consumerId);
  if (consIter != consumerTable.end())
  {
    boost::shared_ptr<DQMConsumerPipe> consPipe = consIter->second;
    bufPtr = consPipe->getDQMEvent();
  }

  // return the event buffer
  return bufPtr;
}

void DQMEventServer::processDQMEvent

(

const DQMEventMsgView &

eventView

)

Processes the specified event.

This includes checking whether the event is allowed to be delivered to consumers based on the prescale and maximum event rate specified in the constructor, checking if any consumers are ready to receive events, checking if any consumers are interested in this specific event, making a local copy of the event, and saving the event to be delivered to consumers.

Definition at line 113 of file DQMEventServer.cc.

References consumerTable, edmNew::copy(), disconnectedConsumerTestCounter_, lat::endl(), DQMEventMsgView::eventNumberAtUpdate(), FDEBUG, lastAcceptedEventTime_, NULL, DQMEventMsgView::size(), source, DQMEventMsgView::startAddress(), target, and DQMEventMsgView::topFolderName().

{
  // check if we are ready to accept another event
  struct timeval now;
  // throttle events that occur more frequently than our max allowed rate
  struct timezone dummyTZ;
  gettimeofday(&now, &dummyTZ);
  double timeDiff = (double) now.tv_sec;
  timeDiff -= (double) lastAcceptedEventTime_.tv_sec;
  timeDiff += ((double) now.tv_usec / 1000000.0);
  timeDiff -= ((double) lastAcceptedEventTime_.tv_usec / 1000000.0);
  //cout << "timeDiff = " << timeDiff <<
  //  ", minTime = " << minTimeBetweenEvents_ << std::endl;
  //if (timeDiff < minTimeBetweenEvents_) {return;}

  // actually do not throttle for now as DQMEvents all come at once
  // for different top level folders and we can miss the second and
  // subsequent top level folders for the same update if we throttle this
  // way, we should include the updateAtEventNumber in deciding
  // we need a buffer that can hold more than one DQMEvent because all
  // FUs are going to send their same folderID DQMEvents at once

  // do nothing if the event is empty
  if (eventView.size() == 0) {return;}

  // loop over the consumers in our list, and for each one check whether
  // it is ready for an event and if it wants this specific event.  If so,
  // create a local copy of the event (if not already done) and pass it
  // to the consumer pipe.
  boost::shared_ptr< vector<char> > bufPtr;
  std::map< uint32, boost::shared_ptr<DQMConsumerPipe> >::const_iterator consIter;
  for (consIter = consumerTable.begin();
       consIter != consumerTable.end();
       consIter++)
  {
    // test if the consumer is ready and wants the event
    boost::shared_ptr<DQMConsumerPipe> consPipe = consIter->second;
    FDEBUG(5) << "Checking if consumer " << consPipe->getConsumerId() <<
      " wants update " << eventView.eventNumberAtUpdate() << 
      " and folder " << eventView.topFolderName() << std::endl;
    
    if (consPipe->isReadyForEvent() &&
        consPipe->wantsDQMEvent(eventView))
    {
      // check if we need to make a local copy of the event
      if (bufPtr.get() == NULL)
      {
        FDEBUG(5) << "Creating a buffer for update " <<
          eventView.eventNumberAtUpdate() << 
          " and folder " << eventView.topFolderName() <<std::endl;

        // create a local buffer of the appropriate size
        boost::shared_ptr< vector<char> >
          tmpBufPtr(new vector<char>(eventView.size()));

        // copy the data to the local buffer
        unsigned char *target = (unsigned char *) &(*tmpBufPtr)[0];
        unsigned char *source = eventView.startAddress();
        int dataSize = eventView.size();
        std::copy(source, source+dataSize, target);

        // switch buffers
        bufPtr.swap(tmpBufPtr); // what happens to the memory and do we need a delete?

        // update the local time stamp for the latest accepted event
        lastAcceptedEventTime_ = now;
      }
      FDEBUG(5) << "Adding DQMevent to consumer pipe for update " <<
          eventView.eventNumberAtUpdate() << 
          " and folder " << eventView.topFolderName() <<std::endl;

      // add the event to the consumer pipe
      consPipe->putDQMEvent(bufPtr);
    }
  }

  // periodically check for disconnected consumers
  disconnectedConsumerTestCounter_++;
  if (disconnectedConsumerTestCounter_ >= 500)
  {
    // reset counter
    disconnectedConsumerTestCounter_ = 0;

    // determine which consumers have disconnected
    std::vector<uint32> disconnectList;
    std::map< uint32, boost::shared_ptr<DQMConsumerPipe> >::const_iterator consIter;
    for (consIter = consumerTable.begin();
         consIter != consumerTable.end();
         consIter++)
    {
      boost::shared_ptr<DQMConsumerPipe> consPipe = consIter->second;
      FDEBUG(5) << "Checking if DQM consumer " << consPipe->getConsumerId() <<
        " has disconnected " << std::endl;
      if (consPipe->isDisconnected())
      {
        disconnectList.push_back(consIter->first);
      }
    }

    // remove disconnected consumers from the consumer table
    std::vector<uint32>::const_iterator listIter;
    for (listIter = disconnectList.begin();
         listIter != disconnectList.end();
         listIter++)
    {
      uint32 consumerId = *listIter;
      consumerTable.erase(consumerId);
    }
  }
}

---

Member Data Documentation

std::map< uint32, boost::shared_ptr<DQMConsumerPipe> > stor::DQMEventServer::consumerTable [private]

Definition at line 57 of file DQMEventServer.h.

Referenced by addConsumer(), clearQueue(), getConsumer(), getConsumerTable(), getDQMEvent(), and processDQMEvent().

int stor::DQMEventServer::disconnectedConsumerTestCounter_ [private]

Definition at line 54 of file DQMEventServer.h.

Referenced by DQMEventServer(), and processDQMEvent().

struct timeval stor::DQMEventServer::lastAcceptedEventTime_ [read, private]

Definition at line 51 of file DQMEventServer.h.

Referenced by DQMEventServer(), and processDQMEvent().

const double DQMEventServer::MAX_ACCEPT_INTERVAL = 86400.0 [static, private]

Initialize the maximum accept interval.

Definition at line 49 of file DQMEventServer.h.

Referenced by DQMEventServer().

double stor::DQMEventServer::minTimeBetweenEvents_ [private]

Definition at line 50 of file DQMEventServer.h.

Referenced by DQMEventServer().

---

The documentation for this class was generated from the following files:

• EventFilter/StorageManager/interface/DQMEventServer.h
• EventFilter/StorageManager/src/DQMEventServer.cc

---