---

stor::DataProcessManager Class Reference

#include <EventFilter/SMProxyServer/interface/DataProcessManager.h>

List of all members.

Public Types
typedef std::vector< char >	Buf
typedef std::map< std::string, bool >	HeaderConsumer_map
typedef std::map< std::string, struct timeval >	LastReqTime_map
typedef std::map< std::string, unsigned int >	RegConsumer_map
enum	STATS_TIME_FRAME { SHORT_TERM = 0, LONG_TERM = 1 }
enum	STATS_TIMING_TYPE { EVENT_FETCH = 10, DQMEVENT_FETCH = 11 }
Public Member Functions
void	addDQMSM2Register (std::string DQMsmURL)
void	addMeasurement (unsigned long size)
void	addSM2Register (std::string smURL)
	DataProcessManager ()
double	getAverageValue (STATS_TIME_FRAME timeFrame=SHORT_TERM, STATS_TIMING_TYPE timingType=EVENT_FETCH, double currentTime=BaseCounter::getCurrentTime())
edm::EventBuffer &	getCommandQueue ()
boost::shared_ptr < DQMEventServer > &	getDQMEventServer ()
boost::shared_ptr < stor::DQMServiceManager > &	getDQMServiceManager ()
double	getDuration (STATS_TIME_FRAME timeFrame=SHORT_TERM, STATS_TIMING_TYPE timingType=EVENT_FETCH, double currentTime=BaseCounter::getCurrentTime())
boost::shared_ptr< EventServer > &	getEventServer ()
boost::shared_ptr < InitMsgCollection > &	getInitMsgCollection ()
double	getSampleCount (STATS_TIME_FRAME timeFrame=SHORT_TERM, STATS_TIMING_TYPE timingType=EVENT_FETCH, double currentTime=BaseCounter::getCurrentTime())
stor::SMPerfStats	getStats ()
bool	haveHeader ()
bool	haveRegWithDQMServer ()
bool	haveRegWithEventServer ()
void	join ()
unsigned long	period4samples ()
unsigned long	receivedDQMevents ()
unsigned long	receivedevents ()
unsigned long	samples ()
void	setArchiveDQM (bool archiveDQM)
void	setArchiveIntervalDQM (int archiveInterval)
void	setCollateDQM (bool collateDQM)
void	setCompressionLevelDQM (int compressionLevelDQM)
void	setConsumerName (std::string s)
void	setDQMConsumerName (std::string s)
void	setDQMEventServer (boost::shared_ptr< DQMEventServer > &es)
void	setEventServer (boost::shared_ptr< EventServer > &es)
void	setFilePrefixDQM (std::string filePrefixDQM)
void	setInitMsgCollection (boost::shared_ptr< InitMsgCollection > &imColl)
void	setMaxDQMEventRequestRate (double rate)
void	setMaxEventRequestRate (double rate)
void	setPeriod4Samples (unsigned long period4samples)
void	setPurgeTimeDQM (int purgeTimeDQM)
void	setReadyTimeDQM (int readyTimeDQM)
void	setSamples (unsigned long num_samples)
void	setUseCompressionDQM (bool useCompressionDQM)
void	start ()
void	stop ()
double	totalvolumemb ()
void	updateMinEventRequestInterval ()
	~DataProcessManager ()
Private Member Functions
bool	getAnyHeaderFromSM ()
void	getDQMEventFromAllSM ()
double	getDQMTime2Wait (std::string smURL)
void	getEventFromAllSM ()
bool	getHeaderFromAllSM ()
bool	getHeaderFromSM (std::string smURL)
bool	getOneDQMEventFromSM (std::string smURL, double &time2wait)
bool	getOneEventFromSM (std::string smURL, double &time2wait)
double	getTime2Wait (std::string smURL)
void	init ()
void	processCommands ()
bool	registerWithAllDQMSM ()
bool	registerWithAllSM ()
int	registerWithDQMSM (std::string smURL)
int	registerWithSM (std::string smURL)
void	setDQMTime2Now (std::string smURL)
void	setTime2Now (std::string smURL)
void	waitBetweenRegTrys ()
Static Private Member Functions
static void	run (DataProcessManager *)
Private Attributes
bool	alreadyRegistered_
bool	alreadyRegisteredDQM_
std::vector< unsigned char >	buf_
edm::EventBuffer *	cmd_q_
unsigned int	consumerId_
std::string	consumerName_
std::string	consumerPriority_
std::string	consumerPSetString_
std::string	consumerTopFolderName_
unsigned int	DQMconsumerId_
std::string	DQMconsumerName_
std::string	DQMconsumerPriority_
std::string	DQMeventpage_
boost::shared_ptr< DQMEventServer >	DQMeventServer_
edm::CPUTimer	dqmFetchTimer_
std::string	DQMregpage_
boost::shared_ptr < stor::DQMServiceManager >	dqmServiceManager_
std::vector< std::string >	DQMsmList_
RegConsumer_map	DQMsmRegMap_
edm::CPUTimer	eventFetchTimer_
std::string	eventpage_
boost::shared_ptr< EventServer >	eventServer_
std::string	headerpage_
bool	headerRefetchRequested_
int	headerRetryInterval_
boost::shared_ptr < InitMsgCollection >	initMsgCollection_
LastReqTime_map	lastDQMReqMap_
LastReqTime_map	lastReqMap_
boost::shared_ptr< ForeverCounter >	ltDQMFetchTimeCounter_
boost::shared_ptr< ForeverCounter >	ltEventFetchTimeCounter_
double	maxEventRequestRate_
boost::shared_ptr< boost::thread >	me_
double	minDQMEventRequestInterval_
double	minEventRequestInterval_
xdata::UnsignedInteger32	period4samples_
stor::SMPerformanceMeter *	pmeter_
unsigned long	receivedDQMEvents_
unsigned long	receivedEvents_
std::string	regpage_
xdata::UnsignedInteger32	samples_
HeaderConsumer_map	smHeaderMap_
std::vector< std::string >	smList_
RegConsumer_map	smRegMap_
stor::SMPerfStats	stats_
boost::shared_ptr < RollingIntervalCounter >	stDQMFetchTimeCounter_
boost::shared_ptr < RollingIntervalCounter >	stEventFetchTimeCounter_
char	subscriptionurl_ [2048]

---

Detailed Description

Definition at line 28 of file DataProcessManager.h.

---

Member Typedef Documentation

typedef std::vector<char> stor::DataProcessManager::Buf

Definition at line 34 of file DataProcessManager.h.

typedef std::map<std::string, bool> stor::DataProcessManager::HeaderConsumer_map

Definition at line 36 of file DataProcessManager.h.

typedef std::map<std::string, struct timeval> stor::DataProcessManager::LastReqTime_map

Definition at line 37 of file DataProcessManager.h.

typedef std::map<std::string, unsigned int> stor::DataProcessManager::RegConsumer_map

Definition at line 35 of file DataProcessManager.h.

---

Member Enumeration Documentation

enum stor::DataProcessManager::STATS_TIME_FRAME

Enumerator:

SHORT_TERM
LONG_TERM

Definition at line 31 of file DataProcessManager.h.

{ SHORT_TERM = 0, LONG_TERM = 1 };

enum stor::DataProcessManager::STATS_TIMING_TYPE

Enumerator:

EVENT_FETCH
DQMEVENT_FETCH

Definition at line 32 of file DataProcessManager.h.

{ EVENT_FETCH = 10, DQMEVENT_FETCH = 11 };

---

Constructor & Destructor Documentation

stor::DataProcessManager::DataProcessManager

(

)

Definition at line 31 of file DataProcessManager.cc.

References init(), and pmeter_.

                                        :
    cmd_q_(edm::getEventBuffer(voidptr_size,50)),
    alreadyRegistered_(false),
    alreadyRegisteredDQM_(false),
    headerRefetchRequested_(false),
    buf_(2000),
    headerRetryInterval_(5),
    dqmServiceManager_(new stor::DQMServiceManager()),
    receivedEvents_(0),
    receivedDQMEvents_(0),
    samples_(100),
    period4samples_(5)
  {
    // for performance measurements
    pmeter_ = new stor::SMPerformanceMeter();
    init();
  } 

stor::DataProcessManager::~DataProcessManager

(

)

Definition at line 49 of file DataProcessManager.cc.

References pmeter_.

  {
    delete pmeter_;
  }

---

Member Function Documentation

void stor::DataProcessManager::addDQMSM2Register

(

std::string

DQMsmURL

)

Definition at line 332 of file DataProcessManager.cc.

References DQMsmList_, DQMsmRegMap_, i, and lastDQMReqMap_.

  {
    // Check if already in the list
    bool alreadyInList = false;
    if(DQMsmList_.size() > 0) {
       for(unsigned int i = 0; i < DQMsmList_.size(); ++i) {
         if(DQMsmURL.compare(DQMsmList_[i]) == 0) {
            alreadyInList = true;
            break;
         }
       }
    }
    if(alreadyInList) return;
    DQMsmList_.push_back(DQMsmURL);
    DQMsmRegMap_.insert(std::make_pair(DQMsmURL,0));
    struct timeval lastRequestTime;
    lastRequestTime.tv_sec = 0;
    lastRequestTime.tv_usec = 0;
    lastDQMReqMap_.insert(std::make_pair(DQMsmURL,lastRequestTime));
  }

void stor::DataProcessManager::addMeasurement

(

unsigned long

size

)

Definition at line 1154 of file DataProcessManager.cc.

References stor::SMPerformanceMeter::addSample(), stor::SMPerformanceMeter::getStats(), pmeter_, and stats_.

Referenced by getOneDQMEventFromSM(), and getOneEventFromSM().

  {
    // for bandwidth performance measurements
    if(pmeter_->addSample(size))
    {
       stats_ = pmeter_->getStats();
    }
  }

void stor::DataProcessManager::addSM2Register

(

std::string

smURL

)

Definition at line 308 of file DataProcessManager.cc.

References i, lastReqMap_, smHeaderMap_, smList_, smRegMap_, and updateMinEventRequestInterval().

  {
    // This smURL is the URN of the StorageManager without the page extension
    // Check if already in the list
    bool alreadyInList = false;
    if(smList_.size() > 0) {
       for(unsigned int i = 0; i < smList_.size(); ++i) {
         if(smURL.compare(smList_[i]) == 0) {
            alreadyInList = true;
            break;
         }
       }
    }
    if(alreadyInList) return;
    smList_.push_back(smURL);
    smRegMap_.insert(std::make_pair(smURL,0));
    smHeaderMap_.insert(std::make_pair(smURL,false));
    struct timeval lastRequestTime;
    lastRequestTime.tv_sec = 0;
    lastRequestTime.tv_usec = 0;
    lastReqMap_.insert(std::make_pair(smURL,lastRequestTime));
    updateMinEventRequestInterval();
  }

bool stor::DataProcessManager::getAnyHeaderFromSM

(

)

[private]

Definition at line 556 of file DataProcessManager.cc.

References getHeaderFromSM(), i, smList_, and smRegMap_.

  {
    // Try the list of SM in order of registration to get one Header
    bool gotOneHeader = false;
    if(smList_.size() > 0) {
       for(unsigned int i = 0; i < smList_.size(); ++i) {
         if(smRegMap_[smList_[i] ] > 0) {
            bool success = getHeaderFromSM(smList_[i]);
            if(success) { // should cleam this up!
              gotOneHeader = true;
              return gotOneHeader;
            }
         }
       }
    } else {
      // this is a problem (but maybe not with non-blocking processing loop)
      return false;
    }
    return gotOneHeader;
  }

double stor::DataProcessManager::getAverageValue	(	STATS_TIME_FRAME	timeFrame = SHORT_TERM,
		STATS_TIMING_TYPE	timingType = EVENT_FETCH,
		double	currentTime = BaseCounter::getCurrentTime()
	)

Definition at line 1185 of file DataProcessManager.cc.

References DQMEVENT_FETCH, ltDQMFetchTimeCounter_, ltEventFetchTimeCounter_, SHORT_TERM, stDQMFetchTimeCounter_, and stEventFetchTimeCounter_.

  {
    if (timeFrame == SHORT_TERM) {
      if (timingType == DQMEVENT_FETCH) {
        return stDQMFetchTimeCounter_->getValueAverage(currentTime);
      }
      else {
        return stEventFetchTimeCounter_->getValueAverage(currentTime);
      }
    }
    else {
      if (timingType == DQMEVENT_FETCH) {
        return ltDQMFetchTimeCounter_->getValueAverage();
      }
      else {
        return ltEventFetchTimeCounter_->getValueAverage();
      }
    }
  }

edm::EventBuffer& stor::DataProcessManager::getCommandQueue

(

)

[inline]

Definition at line 97 of file DataProcessManager.h.

References cmd_q_.

{ return *cmd_q_; }

void stor::DataProcessManager::getDQMEventFromAllSM

(

)

[private]

Definition at line 984 of file DataProcessManager.cc.

References DQMsmRegMap_, getOneDQMEventFromSM(), i, and smList_.

Referenced by processCommands().

  {
    // Try the list of SM in order of registration to get one event
    // so long as we have the header from SM already
    if(smList_.size() > 0) {
      double time2wait = 0.0;
      double sleepTime = 300.0;
      bool gotOneEvent = false;
      bool gotOne = false;
      for(unsigned int i = 0; i < smList_.size(); ++i) {
        if(DQMsmRegMap_[smList_[i] ] > 0) {   // is registered
          gotOne = getOneDQMEventFromSM(smList_[i], time2wait);
          if(gotOne) {
            gotOneEvent = true;
          } else {
            if(time2wait < sleepTime && time2wait >= 0.0) sleepTime = time2wait;
          }
        }
      }
      // check if we need to sleep (to enforce the allowed request rate)
      // we don't want to ping the StorageManager app too often
      // TODO fixme: Cannot sleep for DQM as this is a long time usually
      //             and we block the event request poll if we sleep!
      //             have to find out how to ensure the correct poll rate
      if(!gotOneEvent) {
        //if(sleepTime > 0.0) usleep(static_cast<int>(1000000 * sleepTime));
      }
    }
  }

boost::shared_ptr<DQMEventServer>& stor::DataProcessManager::getDQMEventServer

(

)

[inline]

Definition at line 94 of file DataProcessManager.h.

References DQMeventServer_.

{ return DQMeventServer_; }

boost::shared_ptr<stor::DQMServiceManager>& stor::DataProcessManager::getDQMServiceManager

(

)

[inline]

Definition at line 95 of file DataProcessManager.h.

References dqmServiceManager_.

{ return dqmServiceManager_; }

double stor::DataProcessManager::getDQMTime2Wait

(

std::string

smURL

)

[private]

Definition at line 1014 of file DataProcessManager.cc.

References lastDQMReqMap_, and minDQMEventRequestInterval_.

Referenced by getOneDQMEventFromSM().

  {
    // calculate time since last ping of this SM in seconds
    struct timeval now;
    struct timezone dummyTZ;
    gettimeofday(&now, &dummyTZ);
    double timeDiff = (double) now.tv_sec;
    timeDiff -= (double) lastDQMReqMap_[smURL].tv_sec;
    timeDiff += ((double) now.tv_usec / 1000000.0);
    timeDiff -= ((double) lastDQMReqMap_[smURL].tv_usec / 1000000.0);
    if (timeDiff < minDQMEventRequestInterval_)
    {
      return (minDQMEventRequestInterval_ - timeDiff);
    }
    else
    {
      return 0.0;
    }
  }

double stor::DataProcessManager::getDuration	(	STATS_TIME_FRAME	timeFrame = SHORT_TERM,
		STATS_TIMING_TYPE	timingType = EVENT_FETCH,
		double	currentTime = BaseCounter::getCurrentTime()
	)

Definition at line 1207 of file DataProcessManager.cc.

References DQMEVENT_FETCH, ltDQMFetchTimeCounter_, ltEventFetchTimeCounter_, SHORT_TERM, stDQMFetchTimeCounter_, and stEventFetchTimeCounter_.

  {
    if (timeFrame == SHORT_TERM) {
      if (timingType == DQMEVENT_FETCH) {
        return stDQMFetchTimeCounter_->getDuration(currentTime);
      }
      else {
        return stEventFetchTimeCounter_->getDuration(currentTime);
      }
    }
    else {
      if (timingType == DQMEVENT_FETCH) {
        return ltDQMFetchTimeCounter_->getDuration(currentTime);
      }
      else {
        return ltEventFetchTimeCounter_->getDuration(currentTime);
      }
    }
  }

void stor::DataProcessManager::getEventFromAllSM

(

)

[private]

Definition at line 784 of file DataProcessManager.cc.

References getOneEventFromSM(), haveHeader(), i, smList_, and smRegMap_.

Referenced by processCommands().

  {
    // Try the list of SM in order of registration to get one event
    // so long as we have the header from SM already
    if(smList_.size() > 0 && haveHeader()) {
      double time2wait = 0.0;
      double sleepTime = 300.0;
      bool gotOneEvent = false;
      bool gotOne = false;
      for(unsigned int i = 0; i < smList_.size(); ++i) {
        if(smRegMap_[smList_[i] ] > 0) {   // is registered
          gotOne = getOneEventFromSM(smList_[i], time2wait);
          if(gotOne) {
            gotOneEvent = true;
          } else {
            if(time2wait < sleepTime && time2wait >= 0.0) sleepTime = time2wait;
          }
        }
      }
      // check if we need to sleep (to enforce the allowed request rate)
      // we don't want to ping the StorageManager app too often
      if(!gotOneEvent) {
        if(sleepTime > 0.0) usleep(static_cast<int>(1000000 * sleepTime));
      }
    }
  }

boost::shared_ptr<EventServer>& stor::DataProcessManager::getEventServer

(

)

[inline]

Definition at line 51 of file DataProcessManager.h.

References eventServer_.

{ return eventServer_; }

bool stor::DataProcessManager::getHeaderFromAllSM

(

)

[private]

Definition at line 577 of file DataProcessManager.cc.

References getHeaderFromSM(), i, smHeaderMap_, smList_, and smRegMap_.

Referenced by processCommands().

  {
    // Try the list of SM in order of registration to get one Header from each
    // TODO: how do we get multiple headers if there are more than one?
    bool gotAllHeaders = true;
    if(smList_.size() > 0) {
       for(unsigned int i = 0; i < smList_.size(); ++i) {
         if(smRegMap_[smList_[i] ] > 0) { // is registered
            if(smHeaderMap_[smList_[i] ]) continue; // already got header
            bool success = getHeaderFromSM(smList_[i]);
            if(success) {
              smHeaderMap_[smList_[i] ] = true;
            } else {
              gotAllHeaders = false;
            }
         } else {
           gotAllHeaders = false;
         }
       }
    } else {
      return false;
    }
    return gotAllHeaders;
  }

bool stor::DataProcessManager::getHeaderFromSM

(

std::string

smURL

)

[private]

Definition at line 602 of file DataProcessManager.cc.

References OtherMessageView::bodySize(), TestMuL1L2Filter_cff::cerr, HeaderView::code(), convert(), stor::ReadData::d_, data, lat::endl(), eventServer_, Exception, FDEBUG, stor::func(), Header::HEADER_REQUEST, headerpage_, Header::INIT, Header::INIT_SET, initMsgCollection_, len, OtherMessageBuilder::msgBody(), OtherMessageView::msgBody(), NULL, stor::setopt(), OtherMessageBuilder::size(), InitMsgView::size(), smRegMap_, and OtherMessageBuilder::startAddress().

Referenced by getAnyHeaderFromSM(), and getHeaderFromAllSM().

  {
    // One single try to get a header from this SM URL
    stor::ReadData data;

    data.d_.clear();
    CURL* han = curl_easy_init();
    if(han==0)
    {
      edm::LogError("getHeaderFromSM") << "Could not create curl handle";
      // this is a fatal error isn't it? Are we catching this? TODO
      throw cms::Exception("getHeaderFromSM","DataProcessManager")
          << "Unable to create curl handle\n";
    }
    // set the standard https request options
    std::string url2use = smURL + headerpage_;
    setopt(han,CURLOPT_URL,url2use.c_str());
    setopt(han,CURLOPT_WRITEFUNCTION,func);
    setopt(han,CURLOPT_WRITEDATA,&data);

    // send our consumer ID as part of the header request
    char msgBuff[100];
    OtherMessageBuilder requestMessage(&msgBuff[0], Header::HEADER_REQUEST,
                                       sizeof(char_uint32));
    uint8 *bodyPtr = requestMessage.msgBody();
    convert(smRegMap_[smURL], bodyPtr);
    setopt(han, CURLOPT_POSTFIELDS, requestMessage.startAddress());
    setopt(han, CURLOPT_POSTFIELDSIZE, requestMessage.size());
    struct curl_slist *headers=NULL;
    headers = curl_slist_append(headers, "Content-Type: application/octet-stream");
    headers = curl_slist_append(headers, "Content-Transfer-Encoding: binary");
    setopt(han, CURLOPT_HTTPHEADER, headers);

    // send the HTTP POST, read the reply, and cleanup before going on
    CURLcode messageStatus = curl_easy_perform(han);
    curl_slist_free_all(headers);
    curl_easy_cleanup(han);

    if(messageStatus!=0)
    { 
      cerr << "curl perform failed for header" << endl;
      edm::LogError("getHeaderFromSM") << "curl perform failed for header. "
        << "Could not get header from an already registered Storage Manager"
        << " at " << smURL;
      return false;
    }
    if(data.d_.length() == 0)
    { 
      return false;
    }

    // rely on https transfer string of correct length!
    int len = data.d_.length();
    FDEBUG(9) << "getHeaderFromSM received registry len = " << len << std::endl;

    // check that we've received a valid INIT message
    // or a set of INIT messages.  Save everything that we receive.
    bool addedNewInitMsg = false;
    try
    {
      HeaderView hdrView(&data.d_[0]);
      if (hdrView.code() == Header::INIT)
      {
        InitMsgView initView(&data.d_[0]);
        if (initMsgCollection_->addIfUnique(initView))
        {
          addedNewInitMsg = true;
        }
      }
      else if (hdrView.code() == Header::INIT_SET)
      {
        OtherMessageView otherView(&data.d_[0]);
        bodyPtr = otherView.msgBody();
        uint32 fullSize = otherView.bodySize();
        while ((unsigned int) (bodyPtr-otherView.msgBody()) < fullSize)
        {
          InitMsgView initView(bodyPtr);
          if (initMsgCollection_->addIfUnique(initView))
          {
            addedNewInitMsg = true;
          }
          bodyPtr += initView.size();
        }
      }
      else
      {
        throw cms::Exception("getHeaderFromSM", "DataProcessManager");
      }
    }
    catch (cms::Exception excpt)
    {
      const unsigned int MAX_DUMP_LENGTH = 1000;
      edm::LogError("getHeaderFromSM") << "========================================";
      edm::LogError("getHeaderFromSM") << "Exception decoding the getRegistryData response!";
      if (data.d_.length() <= MAX_DUMP_LENGTH) {
        edm::LogError("getHeaderFromSM") << "Here is the raw text that was returned:";
        edm::LogError("getHeaderFromSM") << data.d_;
      }
      else {
        edm::LogError("getHeaderFromSM") << "Here are the first " << MAX_DUMP_LENGTH <<
          " characters of the raw text that was returned:";
        edm::LogError("getHeaderFromSM") << (data.d_.substr(0, MAX_DUMP_LENGTH));
      }
      edm::LogError("getHeaderFromSM") << "========================================";
      throw excpt;
    }

    // check if any currently connected consumers did not specify
    // an HLT output module label and we now have multiple, different,
    // INIT messages.  If so, we need to complain because the
    // SelectHLTOutput parameter needs to be specified when there
    // is more than one HLT output module (and correspondingly, more
    // than one INIT message)
    if (addedNewInitMsg && eventServer_.get() != NULL &&
        initMsgCollection_->size() > 1)
    {
      std::map< uint32, boost::shared_ptr<ConsumerPipe> > consumerTable = 
        eventServer_->getConsumerTable();
      std::map< uint32, boost::shared_ptr<ConsumerPipe> >::const_iterator 
        consumerIter;
      for (consumerIter = consumerTable.begin();
           consumerIter != consumerTable.end();
           consumerIter++)
      {
        boost::shared_ptr<ConsumerPipe> consPtr = consumerIter->second;

        if (consPtr->getHLTOutputSelection().empty())
        {
          // store a warning message in the consumer pipe to be
          // sent to the consumer at the next opportunity
          std::string errorString;
          errorString.append("ERROR: The configuration for this ");
          errorString.append("consumer does not specify an HLT output ");
          errorString.append("module.\nPlease specify one of the HLT ");
          errorString.append("output modules listed below as the ");
          errorString.append("SelectHLTOutput parameter ");
          errorString.append("in the InputSource configuration.\n");
          errorString.append(initMsgCollection_->getSelectionHelpString());
          errorString.append("\n");
          consPtr->setRegistryWarning(errorString);
        }
      }
    }

    return true;
  }

boost::shared_ptr<InitMsgCollection>& stor::DataProcessManager::getInitMsgCollection

(

)

[inline]

Definition at line 57 of file DataProcessManager.h.

References initMsgCollection_.

{ return initMsgCollection_; }

bool stor::DataProcessManager::getOneDQMEventFromSM	(	std::string	smURL,
		double &	time2wait
	)			[private]

Definition at line 1042 of file DataProcessManager.cc.

References addMeasurement(), buf_, TestMuL1L2Filter_cff::cerr, OtherMessageView::code(), convert(), GenMuonPlsPt100GeV_cfg::cout, stor::ReadData::d_, data, Header::DONE, Header::DQMEVENT_REQUEST, DQMeventpage_, dqmFetchTimer_, dqmServiceManager_, DQMsmRegMap_, lat::endl(), eventFetchTimer_, Exception, FDEBUG, stor::func(), getDQMTime2Wait(), i, len, ltDQMFetchTimeCounter_, OtherMessageBuilder::msgBody(), NULL, edm::CPUTimer::realTime(), receivedDQMEvents_, edm::CPUTimer::reset(), setDQMTime2Now(), stor::setopt(), OtherMessageBuilder::size(), edm::CPUTimer::start(), OtherMessageBuilder::startAddress(), stDQMFetchTimeCounter_, and edm::CPUTimer::stop().

Referenced by getDQMEventFromAllSM().

  {
    // See if we will exceed the request rate, if so just return false
    // Return values: 
    //    true = we have an event; false = no event (whatever reason)
    // time2wait values:
    //    0.0 = we pinged this SM this time; >0 = did not ping, wait this time
    // if every SM returns false we sleep some time
    time2wait = getDQMTime2Wait(smURL);
    if(time2wait > 0.0) {
      return false;
    } else {
      setDQMTime2Now(smURL);
    }

    // One single try to get a event from this SM URL
    stor::ReadData data;

    // start a measurement of how long the HTTP POST takes
    dqmFetchTimer_.stop();
    dqmFetchTimer_.reset();
    dqmFetchTimer_.start();

    data.d_.clear();
    CURL* han = curl_easy_init();
    if(han==0)
    {
      edm::LogError("getOneDQMEventFromSM") << "Could not create curl handle";
      // this is a fatal error isn't it? Are we catching this? TODO
      throw cms::Exception("getOneDQMEventFromSM","DataProcessManager")
          << "Unable to create curl handle\n";
    }
    // set the standard https request options
    std::string url2use = smURL + DQMeventpage_;
    setopt(han,CURLOPT_URL,url2use.c_str());
    setopt(han,CURLOPT_WRITEFUNCTION,stor::func);
    setopt(han,CURLOPT_WRITEDATA,&data);

    // send our consumer ID as part of the event request
    char msgBuff[100];
    OtherMessageBuilder requestMessage(&msgBuff[0], Header::DQMEVENT_REQUEST,
                                       sizeof(char_uint32));
    uint8 *bodyPtr = requestMessage.msgBody();
    convert(DQMsmRegMap_[smURL], bodyPtr);
    setopt(han, CURLOPT_POSTFIELDS, requestMessage.startAddress());
    setopt(han, CURLOPT_POSTFIELDSIZE, requestMessage.size());
    struct curl_slist *headers=NULL;
    headers = curl_slist_append(headers, "Content-Type: application/octet-stream");
    headers = curl_slist_append(headers, "Content-Transfer-Encoding: binary");
    stor::setopt(han, CURLOPT_HTTPHEADER, headers);

    // send the HTTP POST, read the reply, and cleanup before going on
    CURLcode messageStatus = curl_easy_perform(han);
    curl_slist_free_all(headers);
    curl_easy_cleanup(han);

    if(messageStatus!=0)
    { 
      cerr << "curl perform failed for DQM event" << endl;
      edm::LogError("getOneDQMEventFromSM") << "curl perform failed for DQM event. "
        << "Could not get DQMevent from an already registered Storage Manager"
        << " at " << smURL;

      // keep statistics for all HTTP POSTS
      dqmFetchTimer_.stop();
      ltDQMFetchTimeCounter_->addSample(eventFetchTimer_.realTime());
      stDQMFetchTimeCounter_->addSample(eventFetchTimer_.realTime());

      return false;
    }

    // rely on https transfer string of correct length!
    int len = data.d_.length();
    FDEBUG(9) << "getOneDQMEventFromSM received len = " << len << std::endl;
    if(data.d_.length() == 0)
    { 
      // keep statistics for all HTTP POSTS
      dqmFetchTimer_.stop();
      ltDQMFetchTimeCounter_->addSample(eventFetchTimer_.realTime());
      stDQMFetchTimeCounter_->addSample(eventFetchTimer_.realTime());

      return false;
    }

    buf_.resize(len);
    for (int i=0; i<len ; i++) buf_[i] = data.d_[i];

    // keep statistics for all HTTP POSTS
    dqmFetchTimer_.stop();
    ltDQMFetchTimeCounter_->addSample(eventFetchTimer_.realTime());
    stDQMFetchTimeCounter_->addSample(eventFetchTimer_.realTime());

    // first check if done message
    OtherMessageView msgView(&buf_[0]);

    if (msgView.code() == Header::DONE) {
      // TODO fixme:just print message for now
      std::cout << " SM " << smURL << " has halted" << std::endl;
      return false;
    } else {
      DQMEventMsgView dqmEventView(&buf_[0]);
      ++receivedDQMEvents_;
      addMeasurement((unsigned long)data.d_.length());
      if(dqmServiceManager_.get() != NULL) {
          dqmServiceManager_->manageDQMEventMsg(dqmEventView);
          return true;
      }
    }
    return false;
  }

bool stor::DataProcessManager::getOneEventFromSM	(	std::string	smURL,
		double &	time2wait
	)			[private]

Definition at line 839 of file DataProcessManager.cc.

References addMeasurement(), buf_, TestMuL1L2Filter_cff::cerr, HeaderView::code(), OtherMessageView::code(), convert(), GenMuonPlsPt100GeV_cfg::cout, stor::ReadData::d_, data, Header::DONE, lat::endl(), Header::EVENT, Header::EVENT_REQUEST, eventFetchTimer_, eventpage_, eventServer_, Exception, FDEBUG, stor::func(), getTime2Wait(), headerRefetchRequested_, i, initMsgCollection_, len, ltEventFetchTimeCounter_, OtherMessageBuilder::msgBody(), Header::NEW_INIT_AVAILABLE, NULL, edm::CPUTimer::realTime(), receivedEvents_, edm::CPUTimer::reset(), stor::setopt(), setTime2Now(), OtherMessageBuilder::size(), smRegMap_, edm::CPUTimer::start(), OtherMessageBuilder::startAddress(), stEventFetchTimeCounter_, and edm::CPUTimer::stop().

Referenced by getEventFromAllSM().

  {
    // See if we will exceed the request rate, if so just return false
    // Return values: 
    //    true = we have an event; false = no event (whatever reason)
    // time2wait values:
    //    0.0 = we pinged this SM this time; >0 = did not ping, wait this time
    // if every SM returns false we sleep some time
    time2wait = getTime2Wait(smURL);
    if(time2wait > 0.0) {
      return false;
    } else {
      setTime2Now(smURL);
    }

    // One single try to get a event from this SM URL
    stor::ReadData data;

    // start a measurement of how long the HTTP POST takes
    eventFetchTimer_.stop();
    eventFetchTimer_.reset();
    eventFetchTimer_.start();

    data.d_.clear();
    CURL* han = curl_easy_init();
    if(han==0)
    {
      edm::LogError("getOneEventFromSM") << "Could not create curl handle";
      // this is a fatal error isn't it? Are we catching this? TODO
      throw cms::Exception("getOneEventFromSM","DataProcessManager")
          << "Unable to create curl handle\n";
    }
    // set the standard https request options
    std::string url2use = smURL + eventpage_;
    setopt(han,CURLOPT_URL,url2use.c_str());
    setopt(han,CURLOPT_WRITEFUNCTION,stor::func);
    setopt(han,CURLOPT_WRITEDATA,&data);

    // send our consumer ID as part of the event request
    // The event request body consists of the consumerId and the
    // number of INIT messages in our collection.  The latter is used
    // to determine if we need to re-fetch the INIT message collection.
    char msgBuff[100];
    OtherMessageBuilder requestMessage(&msgBuff[0], Header::EVENT_REQUEST,
                                       2 * sizeof(char_uint32));
    uint8 *bodyPtr = requestMessage.msgBody();
    convert(smRegMap_[smURL], bodyPtr);
    bodyPtr += sizeof(char_uint32);
    convert((uint32) initMsgCollection_->size(), bodyPtr);
    setopt(han, CURLOPT_POSTFIELDS, requestMessage.startAddress());
    setopt(han, CURLOPT_POSTFIELDSIZE, requestMessage.size());
    struct curl_slist *headers=NULL;
    headers = curl_slist_append(headers, "Content-Type: application/octet-stream");
    headers = curl_slist_append(headers, "Content-Transfer-Encoding: binary");
    stor::setopt(han, CURLOPT_HTTPHEADER, headers);

    // send the HTTP POST, read the reply, and cleanup before going on
    CURLcode messageStatus = curl_easy_perform(han);
    curl_slist_free_all(headers);
    curl_easy_cleanup(han);

    if(messageStatus!=0)
    { 
      cerr << "curl perform failed for event" << endl;
      edm::LogError("getOneEventFromSM") << "curl perform failed for event. "
        << "Could not get event from an already registered Storage Manager"
        << " at " << smURL;

      // keep statistics for all HTTP POSTS
      eventFetchTimer_.stop();
      ltEventFetchTimeCounter_->addSample(eventFetchTimer_.realTime());
      stEventFetchTimeCounter_->addSample(eventFetchTimer_.realTime());

      return false;
    }

    // rely on https transfer string of correct length!
    int len = data.d_.length();
    FDEBUG(9) << "getOneEventFromSM received len = " << len << std::endl;
    if(data.d_.length() == 0)
    { 
      // keep statistics for all HTTP POSTS
      eventFetchTimer_.stop();
      ltEventFetchTimeCounter_->addSample(eventFetchTimer_.realTime());
      stEventFetchTimeCounter_->addSample(eventFetchTimer_.realTime());

      return false;
    }

    buf_.resize(len);
    for (int i=0; i<len ; i++) buf_[i] = data.d_[i];

    // keep statistics for all HTTP POSTS
    eventFetchTimer_.stop();
    ltEventFetchTimeCounter_->addSample(eventFetchTimer_.realTime());
    stEventFetchTimeCounter_->addSample(eventFetchTimer_.realTime());

    // first check if done message
    OtherMessageView msgView(&buf_[0]);

    if (msgView.code() == Header::DONE) {
      // TODO fixme:just print message for now
      std::cout << " SM " << smURL << " has halted" << std::endl;
      return false;
    } else if (msgView.code() == Header::NEW_INIT_AVAILABLE) {
      std::cout << "Received NEW_INIT_AVAILABLE message" << std::endl;
      headerRefetchRequested_ = true;
      return false;
    } else {
      // 05-Feb-2008, KAB:  catch (and rethrow) any exceptions decoding
      // the event data so that we can display the returned HTML and
      // (hopefully) give the user a hint as to the cause of the problem.
      try {
        HeaderView hdrView(&buf_[0]);
        if (hdrView.code() != Header::EVENT) {
          throw cms::Exception("getOneEventFromSM", "DataProcessManager");
        }
        EventMsgView eventView(&buf_[0]);
        ++receivedEvents_;
        addMeasurement((unsigned long)data.d_.length());
        if(eventServer_.get() != NULL) {
          eventServer_->processEvent(eventView);
          return true;
        }
      }
      catch (cms::Exception excpt) {
        const unsigned int MAX_DUMP_LENGTH = 1000;
        edm::LogError("getOneEventFromSM") << "========================================";
        edm::LogError("getOneEventFromSM") << "Exception decoding the getEventData response!";
        if (data.d_.length() <= MAX_DUMP_LENGTH) {
          edm::LogError("getOneEventFromSM") << "Here is the raw text that was returned:";
          edm::LogError("getOneEventFromSM") << data.d_;
        }
        else {
          edm::LogError("getOneEventFromSM") << "Here are the first " << MAX_DUMP_LENGTH <<
            " characters of the raw text that was returned:";
          edm::LogError("getOneEventFromSM") << (data.d_.substr(0, MAX_DUMP_LENGTH));
        }
        edm::LogError("getOneEventFromSM") << "========================================";
        throw excpt;
      }
    }
    return false;
  }

double stor::DataProcessManager::getSampleCount	(	STATS_TIME_FRAME	timeFrame = SHORT_TERM,
		STATS_TIMING_TYPE	timingType = EVENT_FETCH,
		double	currentTime = BaseCounter::getCurrentTime()
	)

Definition at line 1163 of file DataProcessManager.cc.

References DQMEVENT_FETCH, ltDQMFetchTimeCounter_, ltEventFetchTimeCounter_, SHORT_TERM, stDQMFetchTimeCounter_, and stEventFetchTimeCounter_.

  {
    if (timeFrame == SHORT_TERM) {
      if (timingType == DQMEVENT_FETCH) {
        return stDQMFetchTimeCounter_->getSampleCount(currentTime);
      }
      else {
        return stEventFetchTimeCounter_->getSampleCount(currentTime);
      }
    }
    else {
      if (timingType == DQMEVENT_FETCH) {
        return ltDQMFetchTimeCounter_->getSampleCount();
      }
      else {
        return ltEventFetchTimeCounter_->getSampleCount();
      }
    }
  }

stor::SMPerfStats stor::DataProcessManager::getStats

(

)

[inline]

Definition at line 111 of file DataProcessManager.h.

References stats_.

{ return stats_; }

double stor::DataProcessManager::getTime2Wait

(

std::string

smURL

)

[private]

Definition at line 811 of file DataProcessManager.cc.

References lastReqMap_, and minEventRequestInterval_.

Referenced by getOneEventFromSM().

  {
    // calculate time since last ping of this SM in seconds
    struct timeval now;
    struct timezone dummyTZ;
    gettimeofday(&now, &dummyTZ);
    double timeDiff = (double) now.tv_sec;
    timeDiff -= (double) lastReqMap_[smURL].tv_sec;
    timeDiff += ((double) now.tv_usec / 1000000.0);
    timeDiff -= ((double) lastReqMap_[smURL].tv_usec / 1000000.0);
    if (timeDiff < minEventRequestInterval_)
    {
      return (minEventRequestInterval_ - timeDiff);
    }
    else
    {
      return 0.0;
    }
  }

bool stor::DataProcessManager::haveHeader

(

)

Definition at line 778 of file DataProcessManager.cc.

References initMsgCollection_.

Referenced by getEventFromAllSM(), and processCommands().

  {
    if(initMsgCollection_->size() > 0) return true;
    return false;
  }

bool stor::DataProcessManager::haveRegWithDQMServer

(

)

Definition at line 767 of file DataProcessManager.cc.

References DQMsmList_, DQMsmRegMap_, and i.

  {
    // registered with any of the SM DQM servers
    if(DQMsmList_.size() > 0) {
      for(unsigned int i = 0; i < DQMsmList_.size(); ++i) {
        if(DQMsmRegMap_[DQMsmList_[i] ] > 0) return true;
      }
    }
    return false;
  }

bool stor::DataProcessManager::haveRegWithEventServer

(

)

Definition at line 756 of file DataProcessManager.cc.

References i, smList_, and smRegMap_.

  {
    // registered with any of the SM event servers
    if(smList_.size() > 0) {
      for(unsigned int i = 0; i < smList_.size(); ++i) {
        if(smRegMap_[smList_[i] ] > 0) return true;
      }
    }
    return false;
  }

void stor::DataProcessManager::init

(

void

)

[private]

Definition at line 54 of file DataProcessManager.cc.

References alreadyRegistered_, alreadyRegisteredDQM_, consumerId_, consumerName_, consumerPriority_, consumerTopFolderName_, DQMconsumerId_, DQMconsumerName_, DQMconsumerPriority_, DQMeventpage_, DQMregpage_, DQMsmList_, DQMsmRegMap_, eventpage_, stor::SMPerfStats::fullReset(), headerpage_, headerRefetchRequested_, stor::SMPerformanceMeter::init(), ltDQMFetchTimeCounter_, ltEventFetchTimeCounter_, period4samples_, pmeter_, stor::PROXY_SERVER_NAME(), receivedDQMEvents_, receivedEvents_, regpage_, samples_, setMaxDQMEventRequestRate(), setMaxEventRequestRate(), smHeaderMap_, smList_, smRegMap_, stats_, stDQMFetchTimeCounter_, and stEventFetchTimeCounter_.

Referenced by DataProcessManager().

  {
    regpage_ =  "/registerConsumer";
    DQMregpage_ = "/registerDQMConsumer";
    eventpage_ = "/geteventdata";
    DQMeventpage_ = "/getDQMeventdata";
    headerpage_ = "/getregdata";
    consumerName_ = stor::PROXY_SERVER_NAME;
    //consumerPriority_ = "PushMode"; // this means push mode!
    consumerPriority_ = "Normal";
    DQMconsumerName_ = stor::PROXY_SERVER_NAME;
    //DQMconsumerPriority_ =  "PushMode"; // this means push mode!
    DQMconsumerPriority_ =  "Normal";

    this->setMaxEventRequestRate(10.0); // just a default until set in config action
    consumerId_ = (time(0) & 0xffffff);  // temporary - will get from ES later

    this->setMaxDQMEventRequestRate(0.2); // set in config later
    DQMconsumerId_ = (time(0) & 0xffffff);  // temporary - will get from ES later

    alreadyRegistered_ = false;
    alreadyRegisteredDQM_ = false;
    headerRefetchRequested_ = false;

    smList_.clear();
    smRegMap_.clear();
    smHeaderMap_.clear();
    DQMsmList_.clear();
    DQMsmRegMap_.clear();

    // TODO fixme: only request folders that connected consumers want?
    consumerTopFolderName_ = "*";
    //consumerTopFolderName_ = "C1";
    receivedEvents_ = 0;
    receivedDQMEvents_ = 0;
    pmeter_->init(samples_, period4samples_);
    stats_.fullReset();

    // initialize the counters that we use for statistics
    ltEventFetchTimeCounter_.reset(new ForeverCounter());
    stEventFetchTimeCounter_.reset(new RollingIntervalCounter(180,5,20));
    ltDQMFetchTimeCounter_.reset(new ForeverCounter());
    stDQMFetchTimeCounter_.reset(new RollingIntervalCounter(180,5,20));
  }

void stor::DataProcessManager::join

(

)

Definition at line 189 of file DataProcessManager.cc.

References me_.

  {
    // invoked from a different thread - block until "me_" is done
    if(me_) me_->join();
  }

unsigned long stor::DataProcessManager::period4samples

(

)

[inline]

Definition at line 116 of file DataProcessManager.h.

References stor::SMPerformanceMeter::getPeriod4Samples(), and pmeter_.

{ return pmeter_->getPeriod4Samples(); }

void stor::DataProcessManager::processCommands

(

)

[private]

Definition at line 195 of file DataProcessManager.cc.

References alreadyRegistered_, alreadyRegisteredDQM_, edm::EventBuffer::OperateBuffer< T >::buffer(), cmd_q_, count, dqmServiceManager_, edm::EventBuffer::empty(), getDQMEventFromAllSM(), getEventFromAllSM(), getHeaderFromAllSM(), haveHeader(), headerRefetchRequested_, NULL, registerWithAllDQMSM(), registerWithAllSM(), edm::EventBuffer::OperateBuffer< T >::size(), smHeaderMap_, and waitBetweenRegTrys().

Referenced by run().

  {
    // called with this data process manager's own thread.
    // first register with the SM for each subfarm
    bool doneWithRegistration = false;
    // TODO fixme: improve method of hardcored fixed retries
    unsigned int count = 0; // keep of count of tries and quit after 255
    unsigned int maxcount = 255;
    bool doneWithDQMRegistration = false;
    unsigned int countDQM = 0; // keep of count of tries and quit after 255
    bool alreadysaid = false;
    bool alreadysaidDQM = false;

    //bool gotOneHeader = false;
    bool gotOneHeaderFromAll = false;
    unsigned int countINIT = 0; // keep of count of tries and quit after 255
    bool alreadysaidINIT = false;

    bool DoneWithJob = false;
    while(!DoneWithJob)
    {
      // work loop
      // if a header re-fetch has been requested, reset the header vars
      if (headerRefetchRequested_) {
        headerRefetchRequested_ = false;
        //gotOneHeader = false;
        gotOneHeaderFromAll = false;
        smHeaderMap_.clear();
        countINIT = 0;
      }
      // register as event consumer to all SM senders
      if(!alreadyRegistered_) {
        if(!doneWithRegistration)
        {
          waitBetweenRegTrys();
          bool success = registerWithAllSM();
          if(success) doneWithRegistration = true;
          ++count;
        }
        // TODO fixme: decide what to do after max tries
        if(count >= maxcount) edm::LogInfo("processCommands") << "Could not register with all SM Servers"
           << " after " << maxcount << " tries";
        if(doneWithRegistration && !alreadysaid) {
          edm::LogInfo("processCommands") << "Registered with all SM Event Servers";
          alreadysaid = true;
        }
        if(doneWithRegistration) alreadyRegistered_ = true;
      }
      // now register as DQM consumers
      if(!alreadyRegisteredDQM_) {
        if(!doneWithDQMRegistration)
        {
          waitBetweenRegTrys();
          bool success = registerWithAllDQMSM();
          if(success) doneWithDQMRegistration = true;
          ++countDQM;
        }
        // TODO fixme: decide what to do after max tries
        if(count >= maxcount) edm::LogInfo("processCommands") << "Could not register with all SM DQMEvent Servers"
          << " after " << maxcount << " tries";
        if(doneWithDQMRegistration && !alreadysaidDQM) {
          edm::LogInfo("processCommands") << "Registered with all SM DQMEvent Servers";
          alreadysaidDQM = true;
        }
        if(doneWithDQMRegistration) alreadyRegisteredDQM_ = true;
      }
      // now get one INIT header (product registry) and save it
      // as long as at least one SMsender registered with
      // TODO fixme: use the data member for got header to go across runs
      // With multiple SMs, we need to get a Header from each else that consumer
      // is counted as not initialized
      // TODO how to we get all INIT messages from each SM (and know it!)
      //if(!gotOneHeader)
      if(!gotOneHeaderFromAll)
      {
        waitBetweenRegTrys();
        //bool success = getAnyHeaderFromSM();
        bool success = getHeaderFromAllSM();
        //if(success) gotOneHeader = true;
        if(success) gotOneHeaderFromAll = true;
        ++countINIT;
      }
      if(countINIT >= maxcount) edm::LogInfo("processCommands") << "Could not get product registry!"
          << " after " << maxcount << " tries";
      //if(gotOneHeader && !alreadysaidINIT) {
      if(gotOneHeaderFromAll && !alreadysaidINIT) {
        edm::LogInfo("processCommands") << "Got the product registry";
        alreadysaidINIT = true;
      }
      //if(alreadyRegistered_ && gotOneHeader && haveHeader()) {
      if(alreadyRegistered_ && gotOneHeaderFromAll && haveHeader()) {
        getEventFromAllSM();
      }
      if(alreadyRegisteredDQM_) {
        getDQMEventFromAllSM();
      }

      // check for any commands - empty() does not block
      if(!cmd_q_->empty())
      {
        // the next line blocks until there is an entry in cmd_q
        edm::EventBuffer::ConsumerBuffer cb(*cmd_q_);
        MsgCode mc(cb.buffer(),cb.size());

        if(mc.getCode()==MsgCode::DONE) DoneWithJob = true;
        // right now we will ignore all messages other than DONE
      }

    } // done with process loop   
    edm::LogInfo("processCommands") << "Received done - stopping";
    if(dqmServiceManager_.get() != NULL) dqmServiceManager_->stop();
  }

unsigned long stor::DataProcessManager::receivedDQMevents

(

)

[inline]

Definition at line 110 of file DataProcessManager.h.

References receivedDQMEvents_.

{ return receivedDQMEvents_; }

unsigned long stor::DataProcessManager::receivedevents

(

)

[inline]

Definition at line 109 of file DataProcessManager.h.

References receivedEvents_.

{ return receivedEvents_; }

bool stor::DataProcessManager::registerWithAllDQMSM

(

)

[private]

Definition at line 369 of file DataProcessManager.cc.

References DQMsmList_, DQMsmRegMap_, i, and registerWithDQMSM().

Referenced by processCommands().

  {
    // One try at registering with the SM on each subfarm
    // return true if registered with all SM 
    // Only make one attempt and return so we can make this thread stop
    if(DQMsmList_.size() == 0) return false;
    bool allRegistered = true;
    for(unsigned int i = 0; i < DQMsmList_.size(); ++i) {
      if(DQMsmRegMap_[DQMsmList_[i] ] > 0) continue; // already registered
      int consumerid = registerWithDQMSM(DQMsmList_[i]);
      if(consumerid > 0) DQMsmRegMap_[DQMsmList_[i] ] = consumerid;
      else allRegistered = false;
    }
    return allRegistered;
  }

bool stor::DataProcessManager::registerWithAllSM

(

)

[private]

Definition at line 353 of file DataProcessManager.cc.

References i, registerWithSM(), smList_, and smRegMap_.

Referenced by processCommands().

  {
    // One try at registering with the SM on each subfarm
    // return true if registered with all SM 
    // Only make one attempt and return so we can make this thread stop
    if(smList_.size() == 0) return false;
    bool allRegistered = true;
    for(unsigned int i = 0; i < smList_.size(); ++i) {
      if(smRegMap_[smList_[i] ] > 0) continue; // already registered
      int consumerid = registerWithSM(smList_[i]);
      if(consumerid > 0) smRegMap_[smList_[i] ] = consumerid;
      else allRegistered = false;
    }
    return allRegistered;
  }

int stor::DataProcessManager::registerWithDQMSM

(

std::string

smURL

)

[private]

Definition at line 472 of file DataProcessManager.cc.

References buf_, TestMuL1L2Filter_cff::cerr, consumerTopFolderName_, stor::ReadData::d_, data, DQMconsumerName_, DQMconsumerPriority_, DQMregpage_, lat::endl(), ConsRegResponseBuilder::ES_NOT_READY, Exception, FDEBUG, stor::func(), i, len, NULL, stor::setopt(), ConsRegRequestBuilder::size(), and ConsRegRequestBuilder::startAddress().

Referenced by registerWithAllDQMSM().

  {
    // Use this same registration method for both event data and DQM data
    // return with consumerID or 0 for failure
    stor::ReadData data;

    data.d_.clear();
    CURL* han = curl_easy_init();
    if(han==0)
    {
      edm::LogError("registerWithDQMSM") << "Could not create curl handle";
      // this is a fatal error isn't it? Are we catching this? TODO
      throw cms::Exception("registerWithDQMSM","DataProcessManager")
          << "Unable to create curl handle\n";
    }
    // set the standard https request options
    std::string url2use = smURL + DQMregpage_;
    setopt(han,CURLOPT_URL,url2use.c_str());
    setopt(han,CURLOPT_WRITEFUNCTION,func);
    setopt(han,CURLOPT_WRITEDATA,&data);

    // build the registration request message to send to the storage manager
    const int BUFFER_SIZE = 2000;
    char msgBuff[BUFFER_SIZE];
    ConsRegRequestBuilder requestMessage(msgBuff, BUFFER_SIZE, DQMconsumerName_,
                                         DQMconsumerPriority_, consumerTopFolderName_);

    // add the request message as a https post
    setopt(han, CURLOPT_POSTFIELDS, requestMessage.startAddress());
    setopt(han, CURLOPT_POSTFIELDSIZE, requestMessage.size());
    struct curl_slist *headers=NULL;
    headers = curl_slist_append(headers, "Content-Type: application/octet-stream");
    headers = curl_slist_append(headers, "Content-Transfer-Encoding: binary");
    setopt(han, CURLOPT_HTTPHEADER, headers);

    // send the HTTP POST, read the reply, and cleanup before going on
    CURLcode messageStatus = curl_easy_perform(han);
    curl_slist_free_all(headers);
    curl_easy_cleanup(han);

    if(messageStatus!=0)
    {
      cerr << "curl perform failed for DQM registration" << endl;
      edm::LogError("registerWithDQMSM") << "curl perform failed for registration. "
        << "Could not register with DQM: probably XDAQ not running on Storage Manager"
        << " at " << smURL;
      return 0;
    }
    uint32 registrationStatus = ConsRegResponseBuilder::ES_NOT_READY;
    int consumerId = 0;
    if(data.d_.length() > 0)
    {
      int len = data.d_.length();
      FDEBUG(9) << "registerWithDQMSM received len = " << len << std::endl;
      buf_.resize(len);
      for (int i=0; i<len ; i++) buf_[i] = data.d_[i];

      try {
        ConsRegResponseView respView(&buf_[0]);
        registrationStatus = respView.getStatus();
        consumerId = respView.getConsumerId();
      }
      catch (cms::Exception excpt) {
        const unsigned int MAX_DUMP_LENGTH = 1000;
        edm::LogError("registerWithDQMSM") << "========================================";
        edm::LogError("registerWithDQMSM") << "Exception decoding the registerWithSM response!";
        if (data.d_.length() <= MAX_DUMP_LENGTH) {
          edm::LogError("registerWithDQMSM") << "Here is the raw text that was returned:";
          edm::LogError("registerWithDQMSM") << data.d_;
        }
        else {
          edm::LogError("registerWithDQMSM") << "Here are the first " << MAX_DUMP_LENGTH <<
            " characters of the raw text that was returned:";
          edm::LogError("registerWithDQMSM") << (data.d_.substr(0, MAX_DUMP_LENGTH));
        }
        edm::LogError("registerWithDQMSM") << "========================================";
        return 0;
      }
    }
    if(registrationStatus == ConsRegResponseBuilder::ES_NOT_READY) return 0;
    FDEBUG(5) << "Consumer ID = " << consumerId << endl;
    return consumerId;
  }

int stor::DataProcessManager::registerWithSM

(

std::string

smURL

)

[private]

Definition at line 385 of file DataProcessManager.cc.

References buf_, TestMuL1L2Filter_cff::cerr, consumerName_, consumerPriority_, consumerPSetString_, stor::ReadData::d_, data, lat::endl(), ConsRegResponseBuilder::ES_NOT_READY, eventServer_, Exception, FDEBUG, stor::func(), i, len, NULL, regpage_, stor::setopt(), ConsRegRequestBuilder::size(), and ConsRegRequestBuilder::startAddress().

Referenced by registerWithAllSM().

  {
    // Use this same registration method for both event data and DQM data
    // return with consumerID or 0 for failure
    stor::ReadData data;

    data.d_.clear();
    CURL* han = curl_easy_init();
    if(han==0)
    {
      edm::LogError("registerWithSM") << "Could not create curl handle";
      // this is a fatal error isn't it? Are we catching this? TODO
      throw cms::Exception("registerWithSM","DataProcessManager")
          << "Unable to create curl handle\n";
    }
    // set the standard https request options
    std::string url2use = smURL + regpage_;
    setopt(han,CURLOPT_URL,url2use.c_str());
    setopt(han,CURLOPT_WRITEFUNCTION,func);
    setopt(han,CURLOPT_WRITEDATA,&data);

    // build the registration request message to send to the storage manager
    const int BUFFER_SIZE = 2000;
    char msgBuff[BUFFER_SIZE];
    ConsRegRequestBuilder requestMessage(msgBuff, BUFFER_SIZE, consumerName_,
                                         consumerPriority_, consumerPSetString_);

    // add the request message as a https post
    setopt(han, CURLOPT_POSTFIELDS, requestMessage.startAddress());
    setopt(han, CURLOPT_POSTFIELDSIZE, requestMessage.size());
    struct curl_slist *headers=NULL;
    headers = curl_slist_append(headers, "Content-Type: application/octet-stream");
    headers = curl_slist_append(headers, "Content-Transfer-Encoding: binary");
    setopt(han, CURLOPT_HTTPHEADER, headers);

    // send the HTTP POST, read the reply, and cleanup before going on
    CURLcode messageStatus = curl_easy_perform(han);
    curl_slist_free_all(headers);
    curl_easy_cleanup(han);

    if(messageStatus!=0)
    {
      cerr << "curl perform failed for registration" << endl;
      edm::LogError("registerWithSM") << "curl perform failed for registration. "
        << "Could not register: probably XDAQ not running on Storage Manager"
        << " at " << smURL;
      return 0;
    }
    uint32 registrationStatus = ConsRegResponseBuilder::ES_NOT_READY;
    int consumerId = 0;
    if(data.d_.length() > 0)
    {
      int len = data.d_.length();
      FDEBUG(9) << "registerWithSM received len = " << len << std::endl;
      buf_.resize(len);
      for (int i=0; i<len ; i++) buf_[i] = data.d_[i];

      try {
        ConsRegResponseView respView(&buf_[0]);
        registrationStatus = respView.getStatus();
        consumerId = respView.getConsumerId();
        if (eventServer_.get() != NULL) {
          eventServer_->setStreamSelectionTable(respView.getStreamSelectionTable());
        }
      }
      catch (cms::Exception excpt) {
        const unsigned int MAX_DUMP_LENGTH = 1000;
        edm::LogError("registerWithSM") << "========================================";
        edm::LogError("registerWithSM") << "Exception decoding the registerWithSM response!";
        if (data.d_.length() <= MAX_DUMP_LENGTH) {
          edm::LogError("registerWithSM") << "Here is the raw text that was returned:";
          edm::LogError("registerWithSM") << data.d_;
        }
        else {
          edm::LogError("registerWithSM") << "Here are the first " << MAX_DUMP_LENGTH <<
            " characters of the raw text that was returned:";
          edm::LogError("registerWithSM") << (data.d_.substr(0, MAX_DUMP_LENGTH));
        }
        edm::LogError("registerWithSM") << "========================================";
        return 0;
      }
    }
    if(registrationStatus == ConsRegResponseBuilder::ES_NOT_READY) return 0;
    FDEBUG(5) << "Consumer ID = " << consumerId << endl;
    return consumerId;
  }

void stor::DataProcessManager::run

(

DataProcessManager *

t

)

[static, private]

Definition at line 166 of file DataProcessManager.cc.

References processCommands().

Referenced by start().

  {
    t->processCommands();
  }

unsigned long stor::DataProcessManager::samples

(

)

[inline]

Definition at line 115 of file DataProcessManager.h.

References stor::SMPerformanceMeter::getSetSamples(), and pmeter_.

{ return pmeter_->getSetSamples(); }

void stor::DataProcessManager::setArchiveDQM

(

bool

archiveDQM

)

[inline]

Definition at line 67 of file DataProcessManager.h.

References dqmServiceManager_.

    { dqmServiceManager_->setArchiveDQM(archiveDQM); }

void stor::DataProcessManager::setArchiveIntervalDQM

(

int

archiveInterval

)

[inline]

Definition at line 70 of file DataProcessManager.h.

References dqmServiceManager_.

    { dqmServiceManager_->setArchiveInterval(archiveInterval); }

void stor::DataProcessManager::setCollateDQM

(

bool

collateDQM

)

[inline]

Definition at line 64 of file DataProcessManager.h.

References dqmServiceManager_.

    { dqmServiceManager_->setCollateDQM(collateDQM); }

void stor::DataProcessManager::setCompressionLevelDQM

(

int

compressionLevelDQM

)

[inline]

Definition at line 85 of file DataProcessManager.h.

References dqmServiceManager_.

    { dqmServiceManager_->setCompressionLevel(compressionLevelDQM);}

void stor::DataProcessManager::setConsumerName

(

std::string

s

)

[inline]

Definition at line 99 of file DataProcessManager.h.

References consumerName_.

{ consumerName_ = s; }

void stor::DataProcessManager::setDQMConsumerName

(

std::string

s

)

[inline]

Definition at line 100 of file DataProcessManager.h.

References DQMconsumerName_.

{ DQMconsumerName_ = s; }

void stor::DataProcessManager::setDQMEventServer

(

boost::shared_ptr< DQMEventServer > &

es

)

[inline]

Definition at line 88 of file DataProcessManager.h.

References DQMeventServer_, dqmServiceManager_, and NULL.

    {
      // The auto_ptr still owns the memory after this get()
      if (dqmServiceManager_.get() != NULL) dqmServiceManager_->setDQMEventServer(es);
      DQMeventServer_ = es;
    }

void stor::DataProcessManager::setDQMTime2Now

(

std::string

smURL

)

[private]

Definition at line 1034 of file DataProcessManager.cc.

References lastDQMReqMap_.

Referenced by getOneDQMEventFromSM().

  {
    struct timeval now;
    struct timezone dummyTZ;
    gettimeofday(&now, &dummyTZ);
    lastDQMReqMap_[smURL] = now;
  }

void stor::DataProcessManager::setEventServer

(

boost::shared_ptr< EventServer > &

es

)

[inline]

Definition at line 47 of file DataProcessManager.h.

References eventServer_.

    {
      eventServer_ = es;
    }

void stor::DataProcessManager::setFilePrefixDQM

(

std::string

filePrefixDQM

)

[inline]

Definition at line 79 of file DataProcessManager.h.

References dqmServiceManager_.

    { dqmServiceManager_->setFilePrefix(filePrefixDQM);}

void stor::DataProcessManager::setInitMsgCollection

(

boost::shared_ptr< InitMsgCollection > &

imColl

)

[inline]

Definition at line 53 of file DataProcessManager.h.

References initMsgCollection_.

    {
      initMsgCollection_ = imColl;
    }

void stor::DataProcessManager::setMaxDQMEventRequestRate

(

double

rate

)

Definition at line 119 of file DataProcessManager.cc.

References minDQMEventRequestInterval_.

Referenced by init().

  {
    const double MAX_REQUEST_INTERVAL = 300.0;  // seconds
    if(rate <= 0.0) return; // TODO make sure config is checked!
    if (rate < (1.0 / MAX_REQUEST_INTERVAL)) {
      minDQMEventRequestInterval_ = MAX_REQUEST_INTERVAL;
    }
    else {
      minDQMEventRequestInterval_ = 1.0 / rate;  // seconds
    }
  }

void stor::DataProcessManager::setMaxEventRequestRate

(

double

rate

)

Definition at line 112 of file DataProcessManager.cc.

References maxEventRequestRate_, and updateMinEventRequestInterval().

Referenced by init().

  {
    if(rate <= 0.0) return; // TODO make sure config is checked!
    maxEventRequestRate_ = rate;
    updateMinEventRequestInterval();
  }

void stor::DataProcessManager::setPeriod4Samples

(

unsigned long

period4samples

)

Definition at line 105 of file DataProcessManager.cc.

References period4samples_, pmeter_, and stor::SMPerformanceMeter::setPeriod4Samples().

  { 
    period4samples_ = period4samples;
    pmeter_->setPeriod4Samples(period4samples);
  }

void stor::DataProcessManager::setPurgeTimeDQM

(

int

purgeTimeDQM

)

[inline]

Definition at line 73 of file DataProcessManager.h.

References dqmServiceManager_.

    { dqmServiceManager_->setPurgeTime(purgeTimeDQM);}

void stor::DataProcessManager::setReadyTimeDQM

(

int

readyTimeDQM

)

[inline]

Definition at line 76 of file DataProcessManager.h.

References dqmServiceManager_.

    { dqmServiceManager_->setReadyTime(readyTimeDQM);}

void stor::DataProcessManager::setSamples

(

unsigned long

num_samples

)

Definition at line 99 of file DataProcessManager.cc.

References pmeter_, samples_, and stor::SMPerformanceMeter::setSamples().

  { 
    samples_ = num_samples;
    pmeter_->setSamples(num_samples); 
  }

void stor::DataProcessManager::setTime2Now

(

std::string

smURL

)

[private]

Definition at line 831 of file DataProcessManager.cc.

References lastReqMap_.

Referenced by getOneEventFromSM().

  {
    struct timeval now;
    struct timezone dummyTZ;
    gettimeofday(&now, &dummyTZ);
    lastReqMap_[smURL] = now;
  }

void stor::DataProcessManager::setUseCompressionDQM

(

bool

useCompressionDQM

)

[inline]

Definition at line 82 of file DataProcessManager.h.

References dqmServiceManager_.

    { dqmServiceManager_->setUseCompression(useCompressionDQM);}

void stor::DataProcessManager::start

(

void

)

Definition at line 171 of file DataProcessManager.cc.

References me_, run(), and thread.

  {
    // called from a different thread to start things going

    me_.reset(new boost::thread(boost::bind(DataProcessManager::run,this)));
  }

void stor::DataProcessManager::stop

(

)

Definition at line 178 of file DataProcessManager.cc.

References edm::EventBuffer::OperateBuffer< T >::buffer(), cmd_q_, and edm::EventBuffer::OperateBuffer< T >::commit().

  {
    // called from a different thread - trigger completion to the
    // data process manager loop

    edm::EventBuffer::ProducerBuffer cb(*cmd_q_);
    MsgCode mc(cb.buffer(),MsgCode::DONE);
    mc.setCode(MsgCode::DONE);
    cb.commit(mc.codeSize());
  }

double stor::DataProcessManager::totalvolumemb

(

)

[inline]

Definition at line 117 of file DataProcessManager.h.

References pmeter_, and stor::SMPerformanceMeter::totalvolumemb().

{ return pmeter_->totalvolumemb(); }

void stor::DataProcessManager::updateMinEventRequestInterval

(

)

Definition at line 131 of file DataProcessManager.cc.

References consumerPSetString_, edm::ParameterSet::insert(), maxEventRequestRate_, minEventRequestInterval_, smList_, and edm::ParameterSet::toString().

Referenced by addSM2Register(), and setMaxEventRequestRate().

  {
    const double MAX_REQUEST_INTERVAL = 300.0;  // seconds
    double rate = maxEventRequestRate_;

    if (rate < (1.0 / MAX_REQUEST_INTERVAL)) {
      minEventRequestInterval_ = MAX_REQUEST_INTERVAL;
    }
    else {
      // base the interval on the number of storage managers
      // (so that the requested rate doesn't increase proportionally
      // with number of SMs)
      // We could simply use smCount/rate, but let's be a bit more
      // generous than that so that if one of the SMs isn't queueing
      // events for us, we still get enough rate
      if (smList_.size() <= 1) {
        minEventRequestInterval_ = 1.0 / rate;  // seconds
      }
      else {
        minEventRequestInterval_ = (smList_.size() - 1) / rate;
      }
    }

    // 16-Apr-2008, KAB: set maxEventRequestRate in the parameterSet that
    // we send to the storage manager so that the SM knows how many events
    // to queue for the SMPS.
    edm::ParameterSet ps = ParameterSet();
    Entry maxRateEntry("maxEventRequestRate",
                       (1.0 / minEventRequestInterval_),
                       false);
    ps.insert(true, "maxEventRequestRate", maxRateEntry);
    // TODO fixme: only request event types that are requested by connected consumers?
    consumerPSetString_ = ps.toString();
  }

void stor::DataProcessManager::waitBetweenRegTrys

(

)

[private]

Definition at line 749 of file DataProcessManager.cc.

References headerRetryInterval_.

Referenced by processCommands().

  {
    // for now just a simple wait for a fixed time
    sleep(headerRetryInterval_);
    return;
  }

---

Member Data Documentation

bool stor::DataProcessManager::alreadyRegistered_ [private]

Definition at line 153 of file DataProcessManager.h.

Referenced by init(), and processCommands().

bool stor::DataProcessManager::alreadyRegisteredDQM_ [private]

Definition at line 154 of file DataProcessManager.h.

Referenced by init(), and processCommands().

std::vector<unsigned char> stor::DataProcessManager::buf_ [private]

Definition at line 156 of file DataProcessManager.h.

Referenced by getOneDQMEventFromSM(), getOneEventFromSM(), registerWithDQMSM(), and registerWithSM().

edm::EventBuffer* stor::DataProcessManager::cmd_q_ [private]

Definition at line 151 of file DataProcessManager.h.

Referenced by getCommandQueue(), processCommands(), and stop().

unsigned int stor::DataProcessManager::consumerId_ [private]

Definition at line 176 of file DataProcessManager.h.

Referenced by init().

std::string stor::DataProcessManager::consumerName_ [private]

Definition at line 170 of file DataProcessManager.h.

Referenced by init(), registerWithSM(), and setConsumerName().

std::string stor::DataProcessManager::consumerPriority_ [private]

Definition at line 171 of file DataProcessManager.h.

Referenced by init(), and registerWithSM().

std::string stor::DataProcessManager::consumerPSetString_ [private]

Definition at line 172 of file DataProcessManager.h.

Referenced by registerWithSM(), and updateMinEventRequestInterval().

std::string stor::DataProcessManager::consumerTopFolderName_ [private]

Definition at line 183 of file DataProcessManager.h.

Referenced by init(), and registerWithDQMSM().

unsigned int stor::DataProcessManager::DQMconsumerId_ [private]

Definition at line 179 of file DataProcessManager.h.

Referenced by init().

std::string stor::DataProcessManager::DQMconsumerName_ [private]

Definition at line 181 of file DataProcessManager.h.

Referenced by init(), registerWithDQMSM(), and setDQMConsumerName().

std::string stor::DataProcessManager::DQMconsumerPriority_ [private]

Definition at line 182 of file DataProcessManager.h.

Referenced by init(), and registerWithDQMSM().

std::string stor::DataProcessManager::DQMeventpage_ [private]

Definition at line 165 of file DataProcessManager.h.

Referenced by getOneDQMEventFromSM(), and init().

boost::shared_ptr<DQMEventServer> stor::DataProcessManager::DQMeventServer_ [private]

Definition at line 189 of file DataProcessManager.h.

Referenced by getDQMEventServer(), and setDQMEventServer().

edm::CPUTimer stor::DataProcessManager::dqmFetchTimer_ [private]

Definition at line 205 of file DataProcessManager.h.

Referenced by getOneDQMEventFromSM().

std::string stor::DataProcessManager::DQMregpage_ [private]

Definition at line 166 of file DataProcessManager.h.

Referenced by init(), and registerWithDQMSM().

boost::shared_ptr<stor::DQMServiceManager> stor::DataProcessManager::dqmServiceManager_ [private]

Definition at line 186 of file DataProcessManager.h.

Referenced by getDQMServiceManager(), getOneDQMEventFromSM(), processCommands(), setArchiveDQM(), setArchiveIntervalDQM(), setCollateDQM(), setCompressionLevelDQM(), setDQMEventServer(), setFilePrefixDQM(), setPurgeTimeDQM(), setReadyTimeDQM(), and setUseCompressionDQM().

std::vector<std::string> stor::DataProcessManager::DQMsmList_ [private]

Definition at line 161 of file DataProcessManager.h.

Referenced by addDQMSM2Register(), haveRegWithDQMServer(), init(), and registerWithAllDQMSM().

RegConsumer_map stor::DataProcessManager::DQMsmRegMap_ [private]

Definition at line 162 of file DataProcessManager.h.

Referenced by addDQMSM2Register(), getDQMEventFromAllSM(), getOneDQMEventFromSM(), haveRegWithDQMServer(), init(), and registerWithAllDQMSM().

edm::CPUTimer stor::DataProcessManager::eventFetchTimer_ [private]

Definition at line 204 of file DataProcessManager.h.

Referenced by getOneDQMEventFromSM(), and getOneEventFromSM().

std::string stor::DataProcessManager::eventpage_ [private]

Definition at line 163 of file DataProcessManager.h.

Referenced by getOneEventFromSM(), and init().

boost::shared_ptr<EventServer> stor::DataProcessManager::eventServer_ [private]

Definition at line 188 of file DataProcessManager.h.

Referenced by getEventServer(), getHeaderFromSM(), getOneEventFromSM(), registerWithSM(), and setEventServer().

std::string stor::DataProcessManager::headerpage_ [private]

Definition at line 167 of file DataProcessManager.h.

Referenced by getHeaderFromSM(), and init().

bool stor::DataProcessManager::headerRefetchRequested_ [private]

Definition at line 155 of file DataProcessManager.h.

Referenced by getOneEventFromSM(), init(), and processCommands().

int stor::DataProcessManager::headerRetryInterval_ [private]

Definition at line 173 of file DataProcessManager.h.

Referenced by waitBetweenRegTrys().

boost::shared_ptr<InitMsgCollection> stor::DataProcessManager::initMsgCollection_ [private]

Definition at line 190 of file DataProcessManager.h.

Referenced by getHeaderFromSM(), getInitMsgCollection(), getOneEventFromSM(), haveHeader(), and setInitMsgCollection().

LastReqTime_map stor::DataProcessManager::lastDQMReqMap_ [private]

Definition at line 180 of file DataProcessManager.h.

Referenced by addDQMSM2Register(), getDQMTime2Wait(), and setDQMTime2Now().

LastReqTime_map stor::DataProcessManager::lastReqMap_ [private]

Definition at line 177 of file DataProcessManager.h.

Referenced by addSM2Register(), getTime2Wait(), and setTime2Now().

boost::shared_ptr<ForeverCounter> stor::DataProcessManager::ltDQMFetchTimeCounter_ [private]

Definition at line 208 of file DataProcessManager.h.

Referenced by getAverageValue(), getDuration(), getOneDQMEventFromSM(), getSampleCount(), and init().

boost::shared_ptr<ForeverCounter> stor::DataProcessManager::ltEventFetchTimeCounter_ [private]

Definition at line 206 of file DataProcessManager.h.

Referenced by getAverageValue(), getDuration(), getOneEventFromSM(), getSampleCount(), and init().

double stor::DataProcessManager::maxEventRequestRate_ [private]

Definition at line 174 of file DataProcessManager.h.

Referenced by setMaxEventRequestRate(), and updateMinEventRequestInterval().

boost::shared_ptr<boost::thread> stor::DataProcessManager::me_ [private]

Definition at line 192 of file DataProcessManager.h.

Referenced by join(), and start().

double stor::DataProcessManager::minDQMEventRequestInterval_ [private]

Definition at line 178 of file DataProcessManager.h.

Referenced by getDQMTime2Wait(), and setMaxDQMEventRequestRate().

double stor::DataProcessManager::minEventRequestInterval_ [private]

Definition at line 175 of file DataProcessManager.h.

Referenced by getTime2Wait(), and updateMinEventRequestInterval().

xdata::UnsignedInteger32 stor::DataProcessManager::period4samples_ [private]

Definition at line 200 of file DataProcessManager.h.

Referenced by init(), and setPeriod4Samples().

stor::SMPerformanceMeter* stor::DataProcessManager::pmeter_ [private]

Definition at line 197 of file DataProcessManager.h.

Referenced by addMeasurement(), DataProcessManager(), init(), period4samples(), samples(), setPeriod4Samples(), setSamples(), totalvolumemb(), and ~DataProcessManager().

unsigned long stor::DataProcessManager::receivedDQMEvents_ [private]

Definition at line 196 of file DataProcessManager.h.

Referenced by getOneDQMEventFromSM(), init(), and receivedDQMevents().

unsigned long stor::DataProcessManager::receivedEvents_ [private]

Definition at line 195 of file DataProcessManager.h.

Referenced by getOneEventFromSM(), init(), and receivedevents().

std::string stor::DataProcessManager::regpage_ [private]

Definition at line 164 of file DataProcessManager.h.

Referenced by init(), and registerWithSM().

xdata::UnsignedInteger32 stor::DataProcessManager::samples_ [private]

Definition at line 199 of file DataProcessManager.h.

Referenced by init(), and setSamples().

HeaderConsumer_map stor::DataProcessManager::smHeaderMap_ [private]

Definition at line 160 of file DataProcessManager.h.

Referenced by addSM2Register(), getHeaderFromAllSM(), init(), and processCommands().

std::vector<std::string> stor::DataProcessManager::smList_ [private]

Definition at line 158 of file DataProcessManager.h.

Referenced by addSM2Register(), getAnyHeaderFromSM(), getDQMEventFromAllSM(), getEventFromAllSM(), getHeaderFromAllSM(), haveRegWithEventServer(), init(), registerWithAllSM(), and updateMinEventRequestInterval().

RegConsumer_map stor::DataProcessManager::smRegMap_ [private]

Definition at line 159 of file DataProcessManager.h.

Referenced by addSM2Register(), getAnyHeaderFromSM(), getEventFromAllSM(), getHeaderFromAllSM(), getHeaderFromSM(), getOneEventFromSM(), haveRegWithEventServer(), init(), and registerWithAllSM().

stor::SMPerfStats stor::DataProcessManager::stats_ [private]

Definition at line 201 of file DataProcessManager.h.

Referenced by addMeasurement(), getStats(), and init().

boost::shared_ptr<RollingIntervalCounter> stor::DataProcessManager::stDQMFetchTimeCounter_ [private]

Definition at line 209 of file DataProcessManager.h.

Referenced by getAverageValue(), getDuration(), getOneDQMEventFromSM(), getSampleCount(), and init().

boost::shared_ptr<RollingIntervalCounter> stor::DataProcessManager::stEventFetchTimeCounter_ [private]

Definition at line 207 of file DataProcessManager.h.

Referenced by getAverageValue(), getDuration(), getOneEventFromSM(), getSampleCount(), and init().

char stor::DataProcessManager::subscriptionurl_[2048] [private]

Definition at line 168 of file DataProcessManager.h.

---

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

• EventFilter/SMProxyServer/interface/DataProcessManager.h
• EventFilter/SMProxyServer/src/DataProcessManager.cc

---