---

stor::DQMConsumerPipe Class Reference

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

List of all members.

Public Member Functions
void	clearQueue ()
	DQMConsumerPipe (std::string name, std::string priority, int activeTimeout, int idleTimeout, std::string folderName, std::string hostName, int queueSize)
	DQMConsumerPipe constructor.
uint32	getConsumerId () const
	Returns the consumer ID associated with this pipe.
std::string	getConsumerName ()
boost::shared_ptr< std::vector < char > >	getDQMEvent ()
	Fetches the next event from this consumer pipe.
int	getEvents ()
std::string	getHostName ()
time_t	getLastEventRequestTime ()
int	getPushEventFailures ()
void	initializeSelection ()
	Initializes the event selection for this consumer based on the list of available triggers stored in the specified InitMsgView and the request ParameterSet that was specified in the constructor.
bool	isDisconnected () const
	Tests if the consumer has disconnected.
bool	isIdle () const
	Tests if the consumer is idle (as opposed to active).
bool	isProxyServer () const
bool	isReadyForEvent () const
	Tests if the consumer is ready for an event.
void	putDQMEvent (boost::shared_ptr< std::vector< char > > bufPtr)
	Adds the specified event to this consumer pipe.
void	setPushMode (bool mode)
bool	wantsDQMEvent (DQMEventMsgView const &eventView) const
	Tests if the consumer wants the specified event.
	~DQMConsumerPipe ()
	DQMConsumerPipe destructor.
Private Member Functions
bool	pushEvent ()
Private Attributes
uint32	consumerId_
bool	consumerIsProxyServer_
std::string	consumerName_
std::string	consumerPriority_
std::deque< boost::shared_ptr < std::vector< char > > >	eventQueue_
boost::mutex	eventQueueLock_
int	events_
CURL *	han_
struct curl_slist *	headers_
std::string	hostName_
bool	initializationDone
time_t	lastEventRequestTime_
unsigned int	maxQueueSize_
unsigned int	pushEventFailures_
bool	pushMode_
int	timeToDisconnectedState_
int	timeToIdleState_
std::string	topFolderName_
Static Private Attributes
static uint32	rootId_ = 1
	Initialize the static value for the root consumer id.
static boost::mutex	rootIdLock_
	Initialize the static lock used to control access to the root ID.

---

Detailed Description

Definition at line 42 of file DQMConsumerPipe.h.

---

Constructor & Destructor Documentation

DQMConsumerPipe::DQMConsumerPipe	(	std::string	name,
		std::string	priority,
		int	activeTimeout,
		int	idleTimeout,
		std::string	folderName,
		std::string	hostName,
		int	queueSize
	)

DQMConsumerPipe constructor.

Definition at line 36 of file DQMConsumerPipe.cc.

References consumerId_, consumerIsProxyServer_, consumerName_, consumerPriority_, stor::func(), han_, headers_, initializationDone, lastEventRequestTime_, NULL, pushMode_, rootId_, rootIdLock_, stor::setopt(), timeToDisconnectedState_, and timeToIdleState_.

                                               :
  han_(curl_easy_init()),
  headers_(),
  consumerName_(name),consumerPriority_(priority),
  topFolderName_(folderName),
  hostName_(hostName),
  events_(0),
  pushEventFailures_(0),
  maxQueueSize_(queueSize)
{
  // initialize the time values we use for defining "states"
  timeToIdleState_ = activeTimeout;
  timeToDisconnectedState_ = activeTimeout + idleTimeout;
  lastEventRequestTime_ = time(NULL);
  initializationDone = false;
  pushMode_ = false;
  if(consumerPriority_.compare("PushMode") == 0) pushMode_ = true;

  // determine if we're connected to a proxy server
  consumerIsProxyServer_ = false;
  //if (consumerName_ == PROXY_SERVER_NAME)
  if (consumerName_.find("urn") != std::string::npos &&
      consumerName_.find("xdaq") != std::string::npos &&
      consumerName_.find("pushEventData") != std::string::npos)
  {
    consumerIsProxyServer_ = true;
  }

  // assign the consumer ID
  boost::mutex::scoped_lock scopedLockForRootId(rootIdLock_);
  consumerId_ = rootId_;
  rootId_++;

  if(han_==0)
  {
    edm::LogError("DQMConsumerPipe") << "Could not create curl handle";
    //std::cout << "Could not create curl handle" << std::endl;
    // throw exception here when we can make the SM go to a fail state from
    // another thread
  } else {
    headers_ = curl_slist_append(headers_, "Content-Type: application/octet-stream");
    headers_ = curl_slist_append(headers_, "Content-Transfer-Encoding: binary");
    // Avoid the Expect: 100 continue automatic header that gives a 2 sec delay
    // for pthttps but we don't need the Expect: 100 continue anyway
    headers_ = curl_slist_append(headers_, "Expect:");
    setopt(han_, CURLOPT_HTTPHEADER, headers_);
    setopt(han_, CURLOPT_URL, consumerName_.c_str());
    setopt(han_, CURLOPT_WRITEFUNCTION, func);
    // debug options
    //setopt(han_,CURLOPT_VERBOSE, 1);
    //setopt(han_,CURLOPT_TCP_NODELAY, 1);
  }
}

DQMConsumerPipe::~DQMConsumerPipe

(

)

DQMConsumerPipe destructor.

Definition at line 96 of file DQMConsumerPipe.cc.

References consumerId_, lat::endl(), FDEBUG, han_, and headers_.

{
  FDEBUG(5) << "Executing destructor for DQM consumer pipe with ID = " <<
    consumerId_ << std::endl;
  curl_slist_free_all(headers_);
  curl_easy_cleanup(han_);
}

---

Member Function Documentation

void DQMConsumerPipe::clearQueue

(

)

Definition at line 320 of file DQMConsumerPipe.cc.

References eventQueue_, and eventQueueLock_.

Referenced by getDQMEvent().

{
  boost::mutex::scoped_lock scopedLockForEventQueue(eventQueueLock_);
  eventQueue_.clear();
}

uint32 DQMConsumerPipe::getConsumerId

(

)

const

Returns the consumer ID associated with this pipe.

Definition at line 107 of file DQMConsumerPipe.cc.

References consumerId_.

{
  return consumerId_;
}

std::string stor::DQMConsumerPipe::getConsumerName

(

)

[inline]

Definition at line 63 of file DQMConsumerPipe.h.

References consumerName_.

{ return(consumerName_);}

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

(

)

Fetches the next event from this consumer pipe.

If there are no events in the pipe, an empty shared_ptr will be returned (ptr.get() == NULL).

Definition at line 201 of file DQMConsumerPipe.cc.

References clearQueue(), eventQueue_, eventQueueLock_, events_, isDisconnected(), isIdle(), lastEventRequestTime_, and NULL.

{
  // clear out any stale event(s)
  if (isIdle() || isDisconnected())
  {
    this->clearQueue();
  }

  // fetch the most recent event
  boost::shared_ptr< std::vector<char> > bufPtr;
  {
    boost::mutex::scoped_lock scopedLockForEventQueue(eventQueueLock_);
    if (! eventQueue_.empty())
    {
      bufPtr = eventQueue_.front();
      eventQueue_.pop_front();
      ++events_;
    }
  }

  // update the time of the most recent request
  lastEventRequestTime_ = time(NULL);

  // return the event
  return bufPtr;
}

int stor::DQMConsumerPipe::getEvents

(

)

[inline]

Definition at line 65 of file DQMConsumerPipe.h.

References events_.

{ return(events_);}

std::string stor::DQMConsumerPipe::getHostName

(

)

[inline]

Definition at line 67 of file DQMConsumerPipe.h.

References hostName_.

{ return(hostName_);}

time_t stor::DQMConsumerPipe::getLastEventRequestTime

(

)

[inline]

Definition at line 66 of file DQMConsumerPipe.h.

References lastEventRequestTime_.

{ return(lastEventRequestTime_);}

int stor::DQMConsumerPipe::getPushEventFailures

(

)

[inline]

Definition at line 64 of file DQMConsumerPipe.h.

References pushEventFailures_.

{ return(pushEventFailures_);}

void DQMConsumerPipe::initializeSelection

(

)

Initializes the event selection for this consumer based on the list of available triggers stored in the specified InitMsgView and the request ParameterSet that was specified in the constructor.

Definition at line 117 of file DQMConsumerPipe.cc.

References consumerId_, lat::endl(), FDEBUG, and initializationDone.

{
  FDEBUG(5) << "Initializing DQM consumer pipe, ID = " <<
    consumerId_ << std::endl;

  // no need for initialization yet
  initializationDone = true;

}

bool DQMConsumerPipe::isDisconnected

(

)

const

Tests if the consumer has disconnected.

Definition at line 142 of file DQMConsumerPipe.cc.

References lastEventRequestTime_, NULL, and timeToDisconnectedState_.

Referenced by getDQMEvent().

{
  time_t timeDiff = time(NULL) - lastEventRequestTime_;
  return (timeDiff >= timeToDisconnectedState_);
}

bool DQMConsumerPipe::isIdle

(

)

const

Tests if the consumer is idle (as opposed to active).

The idle state indicates that the consumer is still connected, but it hasn't requested an event in some time.

Definition at line 132 of file DQMConsumerPipe.cc.

References lastEventRequestTime_, NULL, timeToDisconnectedState_, and timeToIdleState_.

Referenced by getDQMEvent().

{
  time_t timeDiff = time(NULL) - lastEventRequestTime_;
  return (timeDiff >= timeToIdleState_ &&
          timeDiff <  timeToDisconnectedState_);
}

bool stor::DQMConsumerPipe::isProxyServer

(

)

const [inline]

Definition at line 57 of file DQMConsumerPipe.h.

References consumerIsProxyServer_.

{ return consumerIsProxyServer_; }

bool DQMConsumerPipe::isReadyForEvent

(

)

const

Tests if the consumer is ready for an event.

Definition at line 151 of file DQMConsumerPipe.cc.

References initializationDone, lastEventRequestTime_, NULL, and timeToIdleState_.

{
  // 13-Oct-2006, KAB - we're not ready if we haven't been initialized
  if (! initializationDone) return false;

  // for now, just test if we are in the active state
  time_t timeDiff = time(NULL) - lastEventRequestTime_;
  return (timeDiff < timeToIdleState_);
}

bool DQMConsumerPipe::pushEvent

(

)

[private]

Definition at line 228 of file DQMConsumerPipe.cc.

References TestMuL1L2Filter_cff::cerr, consumerName_, stor::ReadData::d_, data, lat::endl(), eventQueue_, eventQueueLock_, FDEBUG, stor::func(), han_, headers_, i, len, NULL, stor::setopt(), DQMEventMsgView::size(), and DQMEventMsgView::startAddress().

Referenced by putDQMEvent().

{
  // fetch the most recent event
  boost::shared_ptr< std::vector<char> > bufPtr;
  {
    boost::mutex::scoped_lock scopedLockForEventQueue(eventQueueLock_);
    if (! eventQueue_.empty())
    {
      bufPtr = eventQueue_.front();
      eventQueue_.pop_front();
    }
  }
  if (bufPtr.get() == NULL)
  {
    edm::LogError("pushEvent") << "========================================";
    edm::LogError("pushEvent") << "pushEvent called with empty event queue!";
    return false;
  }

  // push the next event out to a push mode consumer (SMProxyServer)
  FDEBUG(5) << "pushing out DQMevent to " << consumerName_ << std::endl;
  stor::ReadData data;

  data.d_.clear();
  // check if curl handle was obtained (at ctor) if not try again
  if(han_==0)
  {
    han_ = curl_easy_init();
    if(han_==0)
    {
      edm::LogError("pushEvent") << "Could not create curl handle";
      return false;
    }
    headers_ = curl_slist_append(headers_, "Content-Type: application/octet-stream");
    headers_ = curl_slist_append(headers_, "Content-Transfer-Encoding: binary");
    // Avoid the Expect: 100 continue automatic header that gives a 2 sec delay
    headers_ = curl_slist_append(headers_, "Expect:");
    setopt(han_, CURLOPT_HTTPHEADER, headers_);
    setopt(han_, CURLOPT_URL, consumerName_.c_str());
    setopt(han_, CURLOPT_WRITEFUNCTION, func);
  }
  setopt(han_,CURLOPT_WRITEDATA,&data);

  // build the event message
  DQMEventMsgView msgView(&(*bufPtr)[0]);

  // add the request message as a https post
  setopt(han_, CURLOPT_POSTFIELDS, msgView.startAddress());
  setopt(han_, CURLOPT_POSTFIELDSIZE, msgView.size());

  // send the HTTP POST, read the reply
  // explicitly close connection when using pthttps transport or sometimes it hangs
  // because somtimes curl does not see that the connection was closed and tries to reuse it
  setopt(han_,CURLOPT_FORBID_REUSE, 1);
  CURLcode messageStatus = curl_easy_perform(han_);

  if(messageStatus!=0)
  {
    cerr << "curl perform failed for pushDQMEvent" << endl;
    edm::LogError("pushEvent") << "curl perform failed for pushDQMEvent. "
        << "Could not register: probably XDAQ not running on Storage Manager"
        << " at " << consumerName_;
    return false;
  }
  // should really read the message to see if okay (if we had made one!)
  if(data.d_.length() == 0)
  {
    return true;
  } else {
    if(data.d_.length() > 0) {
      std::vector<char> buf(1024);
      int len = data.d_.length();
      buf.resize(len);
      for (int i=0; i<len ; i++) buf[i] = data.d_[i];
      const unsigned int MAX_DUMP_LENGTH = 1000;
      edm::LogError("pushEvent") << "========================================";
      edm::LogError("pushEvent") << "Unexpected pushDQMEvent response!";
      if (data.d_.length() <= MAX_DUMP_LENGTH) {
        edm::LogError("pushEvent") << "Here is the raw text that was returned:";
        edm::LogError("pushEvent") << data.d_;
      }
      else {
        edm::LogError("pushEvent") << "Here are the first " << MAX_DUMP_LENGTH <<
          " characters of the raw text that was returned:";
        edm::LogError("pushEvent") << (data.d_.substr(0, MAX_DUMP_LENGTH));
      }
      edm::LogError("pushEvent") << "========================================";
    }
  }
  return false;
}

void DQMConsumerPipe::putDQMEvent

(

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

bufPtr

)

Adds the specified event to this consumer pipe.

Definition at line 175 of file DQMConsumerPipe.cc.

References eventQueue_, eventQueueLock_, events_, lastEventRequestTime_, maxQueueSize_, NULL, pushEvent(), pushEventFailures_, and pushMode_.

{
  // update the local pointer to the most recent event
  boost::mutex::scoped_lock scopedLockForEventQueue(eventQueueLock_);
  eventQueue_.push_back(bufPtr);
  while (eventQueue_.size() > maxQueueSize_) {
    eventQueue_.pop_front();
  }
  // actually push out DQM data if this is a push mode consumer (SMProxyServer)
  if(pushMode_) {
    bool success = pushEvent();
    // update the time of the most recent successful transaction
    if(!success) ++pushEventFailures_;
    else 
    {
      lastEventRequestTime_ = time(NULL);
      ++events_;
    }
  }
}

void stor::DQMConsumerPipe::setPushMode

(

bool

mode

)

[inline]

Definition at line 61 of file DQMConsumerPipe.h.

References pushMode_.

{ pushMode_ = mode; }

bool DQMConsumerPipe::wantsDQMEvent

(

DQMEventMsgView const &

eventView

)

const

Tests if the consumer wants the specified event.

Definition at line 164 of file DQMConsumerPipe.cc.

References DQMEventMsgView::topFolderName(), and topFolderName_.

{
  // for now, only allow one top folder selection or "*"
  std::string meansEverything = "*";
  if(topFolderName_.compare(meansEverything) == 0) return true;
  else return (topFolderName_.compare(eventView.topFolderName()) == 0);
}

---

Member Data Documentation

uint32 stor::DQMConsumerPipe::consumerId_ [private]

Definition at line 74 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), getConsumerId(), initializeSelection(), and ~DQMConsumerPipe().

bool stor::DQMConsumerPipe::consumerIsProxyServer_ [private]

Definition at line 79 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), and isProxyServer().

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

Definition at line 75 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), getConsumerName(), and pushEvent().

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

Definition at line 76 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe().

std::deque< boost::shared_ptr< std::vector<char> > > stor::DQMConsumerPipe::eventQueue_ [private]

Definition at line 96 of file DQMConsumerPipe.h.

Referenced by clearQueue(), getDQMEvent(), pushEvent(), and putDQMEvent().

boost::mutex stor::DQMConsumerPipe::eventQueueLock_ [private]

Definition at line 100 of file DQMConsumerPipe.h.

Referenced by clearQueue(), getDQMEvent(), pushEvent(), and putDQMEvent().

int stor::DQMConsumerPipe::events_ [private]

Definition at line 80 of file DQMConsumerPipe.h.

Referenced by getDQMEvent(), getEvents(), and putDQMEvent().

CURL* stor::DQMConsumerPipe::han_ [private]

Definition at line 71 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), pushEvent(), and ~DQMConsumerPipe().

struct curl_slist* stor::DQMConsumerPipe::headers_ [read, private]

Definition at line 72 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), pushEvent(), and ~DQMConsumerPipe().

std::string stor::DQMConsumerPipe::hostName_ [private]

Definition at line 78 of file DQMConsumerPipe.h.

Referenced by getHostName().

bool stor::DQMConsumerPipe::initializationDone [private]

Definition at line 88 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), initializeSelection(), and isReadyForEvent().

time_t stor::DQMConsumerPipe::lastEventRequestTime_ [private]

Definition at line 85 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), getDQMEvent(), getLastEventRequestTime(), isDisconnected(), isIdle(), isReadyForEvent(), and putDQMEvent().

unsigned int stor::DQMConsumerPipe::maxQueueSize_ [private]

Definition at line 97 of file DQMConsumerPipe.h.

Referenced by putDQMEvent().

unsigned int stor::DQMConsumerPipe::pushEventFailures_ [private]

Definition at line 93 of file DQMConsumerPipe.h.

Referenced by getPushEventFailures(), and putDQMEvent().

bool stor::DQMConsumerPipe::pushMode_ [private]

Definition at line 91 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), putDQMEvent(), and setPushMode().

uint32 DQMConsumerPipe::rootId_ = 1 [static, private]

Initialize the static value for the root consumer id.

Definition at line 103 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe().

boost::mutex DQMConsumerPipe::rootIdLock_ [static, private]

Initialize the static lock used to control access to the root ID.

Definition at line 104 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe().

int stor::DQMConsumerPipe::timeToDisconnectedState_ [private]

Definition at line 84 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), isDisconnected(), and isIdle().

int stor::DQMConsumerPipe::timeToIdleState_ [private]

Definition at line 83 of file DQMConsumerPipe.h.

Referenced by DQMConsumerPipe(), isIdle(), and isReadyForEvent().

std::string stor::DQMConsumerPipe::topFolderName_ [private]

Definition at line 77 of file DQMConsumerPipe.h.

Referenced by wantsDQMEvent().

---

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

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

---