---

DQMNet Class Reference

#include <DQMServices/Core/interface/DQMNet.h>

Inheritance diagram for DQMNet:

List of all members.

Public Types
typedef std::vector< unsigned char >	DataBlob
typedef std::vector< QValue >	QReports
typedef std::vector< uint32_t >	TagList
typedef std::list< WaitObject >	WaitList
Public Member Functions
void	debug (bool doit)
	Enable or disable verbose debugging.
void	delay (int delay)
	Set the I/O dispatching delay.
	DQMNet (const std::string &appname="")
void	listenToCollector (const std::string &host, int port)
	Tell the network layer to connect to host and port and automatically receive updates from upstream DQM sources.
void	lock (void)
	Acquire a lock on the DQM net layer.
virtual int	receive (DQMStore *store)
virtual void	removeLocalObject (const std::string &name)
void	requestFullUpdates (bool doit)
	Enable or disable requests for full updates.
void	run (void)
	Run the actual I/O processing loop.
void	sendLocalChanges (void)
void	sendScalarAsText (bool doit)
	Enable or disable sending scalar monitoring values as text, rather than their ROOT object values.
void	shutdown (void)
	Stop the network layer and wait it to finish.
void	start (void)
	Start running the network layer in a new thread.
void	startLocalServer (int port)
	Start a server socket for accessing this DQM node remotely.
void	unlock (void)
	Release the lock on the DQM net layer.
virtual void	updateLocalObject (Object &o)
void	updateToCollector (const std::string &host, int port)
	Tell the network layer to connect to host and port and automatically send updates whenever local DQM data changes.
virtual	~DQMNet (void)
Static Public Attributes
static const uint32_t	DQM_FLAG_DEAD = 0x80000000
static const uint32_t	DQM_FLAG_NEW = 0x40000000
static const uint32_t	DQM_FLAG_RECEIVED = 0x20000000
static const uint32_t	DQM_FLAG_REPORT_ERROR = 0x1
static const uint32_t	DQM_FLAG_REPORT_OTHER = 0x4
static const uint32_t	DQM_FLAG_REPORT_WARNING = 0x2
static const uint32_t	DQM_FLAG_SCALAR = 0x8
static const uint32_t	DQM_FLAG_TEXT = 0x10000000
static const uint32_t	DQM_FLAG_ZOMBIE = 0x08000000
static const uint32_t	DQM_MSG_GET_OBJECT = 3
static const uint32_t	DQM_MSG_HELLO = 0
static const uint32_t	DQM_MSG_LIST_OBJECTS = 2
static const uint32_t	DQM_MSG_UPDATE_ME = 1
static const uint32_t	DQM_REPLY_LIST_BEGIN = 101
static const uint32_t	DQM_REPLY_LIST_END = 102
static const uint32_t	DQM_REPLY_NONE = 103
static const uint32_t	DQM_REPLY_OBJECT = 104
static const uint32_t	MAX_PEER_WAITREQS = 128
Protected Member Functions
virtual Peer *	createPeer (lat::Socket *s)=0
bool	extractScalarData (DataBlob &objdata, Object &o)
virtual Object *	findObject (Peer *p, const std::string &name, Peer **owner=0)=0
virtual Peer *	getPeer (lat::Socket *s)=0
std::ostream &	logme (void)
virtual Object *	makeObject (Peer *p, const std::string &name)=0
virtual void	markObjectsDead (Peer *p)=0
virtual void	markObjectsZombies (Peer *p)=0
virtual bool	onMessage (Bucket *msg, Peer *p, unsigned char *data, size_t len)
virtual void	purgeDeadObjects (lat::Time oldobj, lat::Time deadobj)=0
bool	reconstructObject (Object &o)
bool	reinstateObject (DQMStore *store, Object &o)
virtual void	releaseFromWait (Bucket *msg, WaitObject &w, Object *o)
virtual void	removePeer (Peer *p, lat::Socket *s)=0
virtual void	requestFullUpdatesFromPeers (void)=0
virtual void	sendObjectListToPeer (Bucket *msg, bool data, bool all, bool clear)=0
virtual void	sendObjectListToPeers (bool all)=0
void	sendObjectToPeer (Bucket *msg, Object &o, bool data, bool text)
virtual bool	shouldStop (void)
void	updateMask (Peer *p)
	Update the selector mask for a peer based on data queues.
virtual void	updatePeerMasks (void)=0
void	waitForData (Peer *p, const std::string &name, const std::string &info, Peer *owner)
	Queue a request for an object and put a peer into the mode of waiting for object data to appear.
Static Protected Member Functions
static void	copydata (Bucket *b, const void *data, size_t len)
Protected Attributes
bool	debug_
bool	requestFullUpdates_
bool	sendScalarAsText_
Private Member Functions
	DQMNet (const DQMNet &)
bool	losePeer (const char *reason, Peer *peer, lat::IOSelectEvent *event, lat::Error *err=0)
	Handle errors with a peer socket.
bool	onLocalNotify (lat::IOSelectEvent *ev)
	React to notifications from the DQM thread.
bool	onPeerConnect (lat::IOSelectEvent *ev)
	Respond to new connections on the server socket.
bool	onPeerData (lat::IOSelectEvent *ev, Peer *p)
	Handle communication to a particular client.
DQMNet &	operator= (const DQMNet &)
void	releaseFromWait (WaitList::iterator i, Object *o)
void	releaseWaiters (Object *o)
void	requestObject (Peer *p, const char *name, size_t len)
	Queue an object request to the data server.
Static Private Member Functions
static void	discard (Bucket *&b)
Private Attributes
std::string	appname_
pthread_t	communicate_
int	delay_
AutoPeer	downstream_
bool	flush_
pthread_mutex_t	lock_
int	pid_
lat::IOSelector	sel_
lat::InetServerSocket *	server_
sig_atomic_t	shutdown_
AutoPeer	upstream_
lat::Time	version_
WaitList	waiting_
lat::Pipe	wakeup_
Classes
struct	AutoPeer
struct	Bucket
struct	CoreObject
struct	Object
struct	Peer
struct	QValue
struct	WaitObject

---

Detailed Description

Definition at line 21 of file DQMNet.h.

---

Member Typedef Documentation

typedef std::vector<unsigned char> DQMNet::DataBlob

Definition at line 48 of file DQMNet.h.

typedef std::vector<QValue> DQMNet::QReports

Definition at line 52 of file DQMNet.h.

typedef std::vector<uint32_t> DQMNet::TagList

Definition at line 51 of file DQMNet.h.

typedef std::list<WaitObject> DQMNet::WaitList

Definition at line 53 of file DQMNet.h.

---

Constructor & Destructor Documentation

DQMNet::DQMNet

(

const std::string &

appname = ""

)

Definition at line 1053 of file DQMNet.cc.

References lat::IOSelector::attach(), lat::CreateHook(), downstream_, lat::IOChannel::fd(), IORead, DQMNet::AutoPeer::next, O_NONBLOCK, onLocalNotify(), DQMNet::AutoPeer::peer, DQMNet::AutoPeer::port, sel_, lat::Pipe::source(), DQMNet::AutoPeer::update, upstream_, wakeup_, and DQMNet::AutoPeer::warned.

  : debug_ (false),
    sendScalarAsText_ (false),
    requestFullUpdates_ (false),
    appname_ (appname.empty() ? "DQMNet" : appname.c_str()),
    pid_ (getpid()),
    server_ (0),
    version_ (Time::current()),
    communicate_ ((pthread_t) -1),
    shutdown_ (0),
    delay_ (1000),
    flush_ (false)
{
  // Create a pipe for the local DQM to tell the communicator
  // thread that local DQM data has changed and that the peers
  // should be notified.
  fcntl(wakeup_.source()->fd(), F_SETFL, O_RDONLY | O_NONBLOCK);
  sel_.attach(wakeup_.source(), IORead, CreateHook(this, &DQMNet::onLocalNotify));

  // Initialise the upstream and downstream to empty.
  upstream_.peer   = downstream_.peer   = 0;
  upstream_.next   = downstream_.next   = 0;
  upstream_.port   = downstream_.port   = 0;
  upstream_.update = downstream_.update = false;
  upstream_.warned = downstream_.warned = false;
}

DQMNet::~DQMNet

(

void

)

[virtual]

Definition at line 1080 of file DQMNet.cc.

{
  // FIXME
}

DQMNet::DQMNet

(

const DQMNet &

)

[private]

---

Member Function Documentation

void DQMNet::copydata	(	Bucket *	b,
		const void *	data,
		size_t	len
	)			[static, protected]

Definition at line 69 of file DQMNet.cc.

References DQMNet::Bucket::data.

Referenced by onMessage(), releaseFromWait(), DQMBasicNet::requestFullUpdatesFromPeers(), requestObject(), run(), DQMImplNet< DQMNet::Object >::sendObjectListToPeer(), and sendObjectToPeer().

{
  b->data.insert(b->data.end(),
                 (const unsigned char *)data,
                 (const unsigned char *)data + len);
}

virtual Peer* DQMNet::createPeer

(

lat::Socket *

s

)

[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by onPeerConnect(), and run().

void DQMNet::debug

(

bool

doit

)

Enable or disable verbose debugging.

Must be called before calling run() or start().

Definition at line 1088 of file DQMNet.cc.

References debug_.

Referenced by DQMService::DQMService().

{
  debug_ = doit;
}

void DQMNet::delay

(

int

delay

)

Set the I/O dispatching delay.

Must be called before calling run() or start().

Definition at line 1096 of file DQMNet.cc.

References delay_.

{
  delay_ = delay;
}

void DQMNet::discard

(

Bucket *&

b

)

[static, private]

Definition at line 78 of file DQMNet.cc.

References DQMNet::Bucket::next.

Referenced by losePeer(), and onPeerData().

{
  while (b)
  {
    Bucket *next = b->next;
    delete b;
    b = next;
  }
}

bool DQMNet::extractScalarData	(	DataBlob &	objdata,
		Object &	o
	)			[protected]

Definition at line 401 of file DQMNet.cc.

References DQM_FLAG_SCALAR, extractNextObject(), DQMNet::CoreObject::flags, VarParsing::obj, DQMNet::CoreObject::object, DQMNet::Object::rawdata, and s.

Referenced by sendObjectToPeer().

{
  if (! o.flags & DQM_FLAG_SCALAR)
    return false;

  TObject *obj = o.object;
  if (! obj && o.rawdata.size())
  {
    DQMRootBuffer buf(DQMRootBuffer::kRead, o.rawdata.size(), &o.rawdata[0], kFALSE);
    buf.InitMap();
    buf.Reset();
    obj = extractNextObject(buf);
  }

  if (TObjString *ostr = dynamic_cast<TObjString *>(obj))
  {
    const TString &s = ostr->String();
    objdata.insert(objdata.end(),
                   (unsigned char *) s.Data(),
                   (unsigned char *) s.Data() + s.Length());
    return true;
  }

  return false;
}

virtual Object* DQMNet::findObject	(	Peer *	p,
		const std::string &	name,
		Peer **	owner = 0
	)			[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by onMessage(), and run().

virtual Peer* DQMNet::getPeer

(

lat::Socket *

s

)

[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by onPeerData().

void DQMNet::listenToCollector	(	const std::string &	host,
		int	port
	)

Tell the network layer to connect to host and port and automatically receive updates from upstream DQM sources.

Must be called before calling run() or start().

Definition at line 1181 of file DQMNet.cc.

References lat::endl(), DQMNet::AutoPeer::host, logme(), DQMNet::AutoPeer::port, DQMNet::AutoPeer::update, and upstream_.

{
  if (! upstream_.host.empty())
  {
    logme()
      << "ERROR: Already receiving data from another collector at "
      << upstream_.host << ":" << upstream_.port << std::endl;
    return;
  }

  upstream_.update = false;
  upstream_.host = host;
  upstream_.port = port;
}

void DQMNet::lock

(

void

)

Acquire a lock on the DQM net layer.

Definition at line 1221 of file DQMNet.cc.

References communicate_, and lock_.

Referenced by DQMService::flush(), onMessage(), DQMBasicNet::receive(), and run().

{
  if (communicate_ != (pthread_t) -1)
    pthread_mutex_lock(&lock_);
}

std::ostream & DQMNet::logme

(

void

)

[protected]

Definition at line 60 of file DQMNet.cc.

References appname_, TestMuL1L2Filter_cff::cerr, and pid_.

Referenced by listenToCollector(), losePeer(), onLocalNotify(), onMessage(), onPeerConnect(), onPeerData(), DQMImplNet< DQMNet::Object >::purgeDeadObjects(), receive(), reconstructObject(), reinstateObject(), removeLocalObject(), DQMImplNet< DQMNet::Object >::requestFullUpdatesFromPeers(), run(), DQMImplNet< DQMNet::Object >::sendObjectListToPeers(), start(), startLocalServer(), updateLocalObject(), updateMask(), and updateToCollector().

{
  return std::cerr
    << Time::current().format(true, "%Y-%m-%d %H:%M:%S")
    << " " << appname_ << "[" << pid_ << "]: ";
}

bool DQMNet::losePeer	(	const char *	reason,
		Peer *	peer,
		lat::IOSelectEvent *	event,
		lat::Error *	err = 0
	)			[private]

Handle errors with a peer socket.

Zaps the socket send queue, the socket itself, detaches the socket from the selector, and purges any pending wait requests linked to the socket.

Definition at line 93 of file DQMNet.cc.

References DQMNet::Peer::automatic, lat::Socket::close(), lat::IOSelector::detach(), discard(), e, lat::endl(), lat::Error::explain(), i, logme(), DQMNet::AutoPeer::peer, DQMNet::Peer::peeraddr, removePeer(), s, sel_, DQMNet::Peer::sendq, DQMNet::Peer::socket, lat::IOSelectEvent::source, and waiting_.

Referenced by onPeerData(), and updateMask().

{
  if (reason)
    logme ()
      << reason << peer->peeraddr
      << (err ? "; error was: " + err->explain() : std::string(""))
      << std::endl;

  Socket *s = peer->socket;

  for (WaitList::iterator i = waiting_.begin(), e = waiting_.end(); i != e; )
    if (i->peer == peer)
      waiting_.erase(i++);
    else
      ++i;

  if (ev)
    ev->source = 0;

  discard(peer->sendq);
  if (peer->automatic)
    peer->automatic->peer = 0;

  sel_.detach (s);
  s->close();
  removePeer (peer, s);
  delete s;
  return true;
}

virtual Object* DQMNet::makeObject	(	Peer *	p,
		const std::string &	name
	)			[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by onMessage().

virtual void DQMNet::markObjectsDead

(

Peer *

p

)

[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by onMessage().

virtual void DQMNet::markObjectsZombies

(

Peer *

p

)

[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by onMessage().

bool DQMNet::onLocalNotify

(

lat::IOSelectEvent *

ev

)

[private]

React to notifications from the DQM thread.

This is a simple message to tell this thread to wake up and send unsollicited updates to the peers when new DQM data appears. We don't send the updates here, but just set a flag to tell the main event pump to send a notification later. This avoids sending unnecessarily frequent DQM object updates.

Definition at line 992 of file DQMNet.cc.

References e, lat::endl(), lat::SysErr::ErrTryAgain, lat::Error::explain(), flush_, logme(), lat::Error::next(), lat::SystemError::portable(), lat::IOChannel::read(), and lat::IOSelectEvent::source.

Referenced by DQMNet().

{
  // Discard the data in the pipe, we care only about the wakeup.
  try
  {
    IOSize sz;
    unsigned char buf [1024];
    while ((sz = ev->source->read(buf, sizeof(buf))))
      ;
  }
  catch (Error &e)
  {
    SystemError *next = dynamic_cast<SystemError *>(e.next());
    if (next && next->portable() == SysErr::ErrTryAgain)
      ; // Ignore it
    else
      logme()
        << "WARNING: error reading from notification pipe: "
        << e.explain() << std::endl;
  }

  // Tell the main event pump to send an update in a little while.
  flush_ = true;

  // We are never done, always keep going.
  return false;
}

bool DQMNet::onMessage	(	Bucket *	msg,
		Peer *	p,
		unsigned char *	data,
		size_t	len
	)			[protected, virtual]

Definition at line 471 of file DQMNet.cc.

References copydata(), DQMNet::Bucket::data, debug_, DQM_FLAG_DEAD, DQM_FLAG_NEW, DQM_FLAG_RECEIVED, DQM_MSG_GET_OBJECT, DQM_MSG_LIST_OBJECTS, DQM_MSG_UPDATE_ME, DQM_REPLY_LIST_BEGIN, DQM_REPLY_LIST_END, DQM_REPLY_NONE, DQM_REPLY_OBJECT, lat::endl(), findObject(), DQMNet::CoreObject::flags, flags, flush_, full, DQMNet::Object::lastreq, lock(), logme(), makeObject(), markObjectsDead(), markObjectsZombies(), name, DQMNet::CoreObject::object, DQMNet::Peer::peeraddr, DQMNet::Object::rawdata, DQMNet::CoreObject::reference, releaseWaiters(), requestFullUpdates_, requestFullUpdatesFromPeers(), requestObject(), sendObjectListToPeer(), sendObjectToPeer(), sendScalarAsText_, DQMNet::Peer::source, DQMNet::CoreObject::tags, unlock(), DQMNet::Peer::update, DQMNet::Peer::updatefull, DQMNet::Peer::updates, DQMNet::CoreObject::version, and waitForData().

Referenced by onPeerData().

{
  // Decode and process this message.
  uint32_t type;
  memcpy (&type, data + sizeof(uint32_t), sizeof (type));
  switch (type)
  {
  case DQM_MSG_UPDATE_ME:
    {
      if (len != 3*sizeof(uint32_t))
      {
        logme()
          << "ERROR: corrupt 'UPDATE_ME' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      // Get the update status: whether this is a full update.
      uint32_t full;
      memcpy(&full, data + 2*sizeof(uint32_t), sizeof(uint32_t));

      if (debug_)
        logme()
          << "DEBUG: received message 'UPDATE ME' from peer "
          << p->peeraddr << ", full = " << full << std::endl;

      p->update = true;
      p->updatefull = full;

      if (full && ! requestFullUpdates_)
      {
        if (debug_)
          logme()
            << "WARNING: forcing full update request mode on due to "
            << "request from " << p->peeraddr << std::endl;
        requestFullUpdates_ = true;
        requestFullUpdatesFromPeers();
      }
    }
    return true;

  case DQM_MSG_LIST_OBJECTS:
    {
      if (debug_)
        logme()
          << "DEBUG: received message 'LIST OBJECTS' from peer "
          << p->peeraddr << std::endl;

      // Send over current status: list of known objects.
      lock();
      sendObjectListToPeer(msg, p->updatefull, true, false);
      unlock();
    }
    return true;

  case DQM_MSG_GET_OBJECT:
    {
      if (debug_)
        logme()
          << "DEBUG: received message 'GET OBJECT' from peer "
          << p->peeraddr << std::endl;

      if (len < 3*sizeof(uint32_t))
      {
        logme()
          << "ERROR: corrupt 'GET IMAGE' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      uint32_t namelen;
      memcpy (&namelen, data + 2*sizeof(uint32_t), sizeof(namelen));
      if (len != 3*sizeof(uint32_t) + namelen)
      {
        logme()
          << "ERROR: corrupt 'GET OBJECT' message of length " << len
          << " from peer " << p->peeraddr
          << ", expected length " << (3*sizeof(uint32_t))
          << " + " << namelen << std::endl;
        return false;
      }

      lock();
      std::string name ((char *) data + 3*sizeof(uint32_t), namelen);
      Peer *owner = 0;
      Object *o = findObject(0, name, &owner);
      if (o)
      {
        o->lastreq = Time::current();
        if (o->rawdata.empty())
          waitForData(p, name, "", owner);
        else
          sendObjectToPeer(msg, *o, true, sendScalarAsText_);
      }
      else
      {
        uint32_t words [3];
        words[0] = sizeof(words) + name.size();
        words[1] = DQM_REPLY_NONE;
        words[2] = name.size();

        msg->data.reserve(msg->data.size() + words[0]);
        copydata(msg, &words[0], sizeof(words));
        copydata(msg, &name[0], name.size());
      }
      unlock();
    }
    return true;

  case DQM_REPLY_LIST_BEGIN:
    {
      if (len != 4*sizeof(uint32_t))
      {
        logme()
          << "ERROR: corrupt 'LIST BEGIN' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      if (debug_)
        logme()
          << "DEBUG: received message 'LIST BEGIN' from "
          << p->peeraddr << std::endl;

      // Get the update status: whether this is a full update.
      uint32_t flags;
      memcpy(&flags, data + 3*sizeof(uint32_t), sizeof(uint32_t));

      // If we are about to receive a full list of objects, flag all
      // objects dead.  Subsequent object notifications will undo this
      // for the live objects.  This tells us the object has been
      // removed, but we can keep making it available for a while if
      // there continues to be interest in it.
      if (flags)
      {
        lock();
        markObjectsZombies(p);
        unlock();
      }
    }
    return true;

  case DQM_REPLY_LIST_END:
    {
      if (len != 4*sizeof(uint32_t))
      {
        logme()
          << "ERROR: corrupt 'LIST END' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      // Get the update status: whether this is a full update.
      uint32_t flags;
      memcpy(&flags, data + 3*sizeof(uint32_t), sizeof(uint32_t));

      // If we received a full list of objects, flag all zombie objects
      // now dead. We need to do this in two stages in case we receive
      // updates in many parts, and end up sending updates to others in
      // between; this avoids us lying live objects are dead.
      if (flags)
      {
        lock();
        markObjectsDead(p);
        unlock();
      }

      if (debug_)
        logme()
          << "DEBUG: received message 'LIST END' from "
          << p->peeraddr << std::endl;

      // Indicate we have received another update from this peer.
      // Also indicate we should flush to our clients.
      flush_ = true;
      p->updates++;
    }
    return true;

  case DQM_REPLY_OBJECT:
    {
      uint32_t words[8];
      if (len < sizeof(words))
      {
        logme()
          << "ERROR: corrupt 'OBJECT' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      memcpy (&words[0], data, sizeof(words));
      uint32_t &namelen = words[5];
      uint32_t &taglen = words[6];
      uint32_t &datalen = words[7];

      if (len != sizeof(words) + namelen + taglen*sizeof(uint32_t) + datalen)
      {
        logme()
          << "ERROR: corrupt 'OBJECT' message of length " << len
          << " from peer " << p->peeraddr
          << ", expected length " << sizeof(words)
          << " + " << namelen
          << " + " << (taglen*sizeof(uint32_t))
          << " + " << datalen
          << std::endl;
        return false;
      }

      unsigned char *namedata = data + sizeof(words);
      unsigned char *tagdata = namedata + namelen;
      unsigned char *objdata = tagdata + taglen*sizeof(uint32_t);
      unsigned char *enddata = objdata + datalen;
      std::string name ((char *) namedata, namelen);
      assert (enddata == data + len);

      if (debug_)
        logme()
          << "DEBUG: received message 'OBJECT " << name
          << "' from " << p->peeraddr << std::endl;

      // Mark the peer as a known object source.
      p->source = true;

      // Initialise or update an object entry.
      lock();
      Object *o = findObject(p, name);
      if (! o)
        o = makeObject(p, name);

      o->flags = words[2] | DQM_FLAG_NEW | DQM_FLAG_RECEIVED;
      o->version = ((uint64_t) words[4] << 32 | words[3]);
      o->tags.clear();
      o->tags.insert(o->tags.end(), (uint32_t *) tagdata, (uint32_t *) objdata);
      o->rawdata.clear();
      o->rawdata.insert (o->rawdata.end(), objdata, enddata);

      bool hadobject = (o->object != 0);
      delete o->object;
      o->object = 0;
      delete o->reference;
      o->reference = 0;

      // If we had an object for this one already and this is a list
      // update without data, issue an immediate data get request.
      if (hadobject && ! datalen)
        requestObject(p, (namelen ? &name[0] : 0), namelen);

      // If we have the object data, release from wait.
      if (datalen)
        releaseWaiters(o);
      unlock();
    }
    return true;

  case DQM_REPLY_NONE:
    {
      uint32_t words[3];
      if (len < sizeof(words))
      {
        logme()
          << "ERROR: corrupt 'NONE' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      memcpy (&words[0], data, sizeof(words));
      uint32_t &namelen = words[2];

      if (len != sizeof(words) + namelen)
      {
        logme()
          << "ERROR: corrupt 'NONE' message of length " << len
          << " from peer " << p->peeraddr
          << ", expected length " << sizeof(words)
          << " + " << namelen << std::endl;
        return false;
      }

      unsigned char *namedata = data + sizeof(words);
      unsigned char *enddata = namedata + namelen;
      std::string name ((char *) namedata, namelen);
      assert (enddata == data + len);

      if (debug_)
        logme()
          << "DEBUG: received message 'NONE " << name
          << "' from " << p->peeraddr << std::endl;

      // Mark the peer as a known object source.
      p->source = true;

      // If this was a known object, update its entry.
      lock();
      Object *o = findObject(p, name);
      if (o)
        o->flags |= DQM_FLAG_DEAD;

      // If someone was waiting for this, let them go.
      releaseWaiters(o);
      unlock();
    }
    return true;

  default:
    logme()
      << "ERROR: unrecognised message of length " << len
      << " and type " << type << " from peer " << p->peeraddr
      << std::endl;
    return false;
  }
}

bool DQMNet::onPeerConnect

(

lat::IOSelectEvent *

ev

)

[private]

Respond to new connections on the server socket.

Accepts the connection and creates a new socket for the peer, and sets it up for further communication. Returns false always to tell the IOSelector to keep processing events for the server socket.

Definition at line 952 of file DQMNet.cc.

References lat::Socket::accept(), arg, lat::IOSelector::attach(), lat::CreateHook(), createPeer(), debug_, lat::endl(), lat::InetAddress::hostname(), IORead, IOUrgent, lat::IOChannel::isBlocking(), logme(), onPeerData(), p, lat::InetAddress::port(), s, sel_, server_, and lat::IOSelectEvent::source.

Referenced by startLocalServer().

{
  // Recover the server socket.
  assert (ev->source == server_);

  // Accept the connection.
  Socket *s = server_->accept();
  assert (s);
  assert (! s->isBlocking());

  // Record it to our list of peers.
  Peer *p = createPeer(s);
  InetAddress peeraddr = ((InetSocket *) s)->peername();
  InetAddress myaddr = ((InetSocket *) s)->sockname();
  p->peeraddr = StringFormat("%1:%2")
                .arg(peeraddr.hostname())
                .arg(peeraddr.port());
  p->mask = IORead|IOUrgent;
  p->socket = s;

  // Report the new connection.
  if (debug_)
    logme()
      << "INFO: new peer " << p->peeraddr << " is now connected to "
      << myaddr.hostname() << ":" << myaddr.port() << std::endl;

  // Attach it to the listener.
  sel_.attach(s, p->mask, CreateHook(this, &DQMNet::onPeerData, p));

  // We are never done.
  return false;
}

bool DQMNet::onPeerData	(	lat::IOSelectEvent *	ev,
		Peer *	p
	)			[private]

Handle communication to a particular client.

Definition at line 786 of file DQMNet.cc.

References DQMNet::Peer::automatic, b, data, DQMNet::Bucket::data, debug_, discard(), e, lat::endl(), lat::SysErr::ErrTryAgain, lat::IOSelectEvent::events, getPeer(), DQMNet::Peer::incoming, IORead, IOUrgent, IOWrite, len, logme(), losePeer(), DQMNet::Peer::mask, MAX_PEER_WAITREQS, MESSAGE_SIZE_LIMIT, alivecheck_mergeAndRegister::msg, DQMNet::Bucket::next, lat::Error::next(), old, onMessage(), DQMNet::Peer::peeraddr, lat::SystemError::portable(), lat::IOChannel::read(), sel_, DQMNet::Peer::sendpos, DQMNet::Peer::sendq, lat::IOSelector::setMask(), DQMNet::Peer::socket, SOCKET_READ_GROWTH, SOCKET_READ_SIZE, lat::IOSelectEvent::source, TrackValidation_HighPurity_cff::valid, DQMNet::Peer::waiting, and lat::IOChannel::write().

Referenced by onPeerConnect(), and run().

{
  assert (getPeer(dynamic_cast<Socket *> (ev->source)) == p);

  // If there is a problem with the peer socket, discard the peer
  // and tell the selector to stop prcessing events for it.  If
  // this is a server connection, we will eventually recreate
  // everything if/when the data server comes back.
  if (ev->events & IOUrgent)
  {
    if (p->automatic)
    {
      logme()
        << "WARNING: connection to the DQM server at " << p->peeraddr
        << " lost (will attempt to reconnect in 15 seconds)\n";
      return losePeer(0, p, ev);
    }
    else
      return losePeer("WARNING: lost peer connection ", p, ev);
  }

  // If we can write to the peer socket, pump whatever we can into it.
  if (ev->events & IOWrite)
  {
    while (Bucket *b = p->sendq)
    {
      IOSize len = b->data.size() - p->sendpos;
      const void *data = (len ? (const void *)&b->data[p->sendpos]
                          : (const void *)&data);
      IOSize done;

      try
      {
        done = (len ? ev->source->write (data, len) : 0);
        if (debug_ && len)
          logme()
            << "DEBUG: sent " << done << " bytes to peer "
            << p->peeraddr << std::endl;
      }
      catch (Error &e)
      {
        return losePeer("WARNING: unable to write to peer ",
                        p, ev, &e);
      }

      p->sendpos += done;
      if (p->sendpos == b->data.size())
      {
        Bucket *old = p->sendq;
        p->sendq = old->next;
        p->sendpos = 0;
        old->next = 0;
        discard(old);
      }

      if (! done && len)
        // Cannot write any more.
        break;
    }
  }

  // If there is data to be read from the peer, first receive what we
  // can get out the socket, the process all complete requests.
  if (ev->events & IORead)
  {
    // First build up the incoming buffer of data in the socket.
    // Remember the last size returned by the socket; we need
    // it to determine if the remote end closed the connection.
    IOSize sz;
    try
    {
      std::vector<unsigned char> buf(SOCKET_READ_SIZE);
      do
        if ((sz = ev->source->read(&buf[0], buf.size())))
        {
          if (debug_)
            logme()
              << "DEBUG: received " << sz << " bytes from peer "
              << p->peeraddr << std::endl;
          DataBlob &data = p->incoming;
          if (data.capacity () < data.size () + sz)
            data.reserve (data.size() + SOCKET_READ_GROWTH);
          data.insert (data.end(), &buf[0], &buf[0] + sz);
        }
      while (sz == sizeof (buf));
    }
    catch (Error &e)
    {
      SystemError *next = dynamic_cast<SystemError *>(e.next());
      if (next && next->portable() == SysErr::ErrTryAgain)
        sz = 1; // Ignore it, and fake no end of data.
      else
        // Houston we have a problem.
        return losePeer("WARNING: failed to read from peer ",
                        p, ev, &e);
    }

    // Process fully received messages as long as we can.
    size_t consumed = 0;
    DataBlob &data = p->incoming;
    while (data.size()-consumed >= sizeof(uint32_t)
           && p->waiting < MAX_PEER_WAITREQS)
    {
      uint32_t msglen;
      memcpy (&msglen, &data[0]+consumed, sizeof(msglen));

      if (msglen >= MESSAGE_SIZE_LIMIT)
        return losePeer("WARNING: excessively large message from ", p, ev);

      if (data.size()-consumed >= msglen)
      {
        bool valid = true;
        if (msglen < 2*sizeof(uint32_t))
        {
          logme()
            << "ERROR: corrupt peer message of length " << msglen
            << " from peer " << p->peeraddr << std::endl;
          valid = false;
        }
        else
        {
          // Decode and process this message.
          Bucket msg;
          msg.next = 0;
          valid = onMessage(&msg, p, &data[0]+consumed, msglen);

          // If we created a response, chain it to the write queue.
          if (! msg.data.empty())
          {
            Bucket **prev = &p->sendq;
            while (*prev)
               prev = &(*prev)->next;

            *prev = new Bucket;
            (*prev)->next = 0;
            (*prev)->data.swap(msg.data);
          }
        }

        if (! valid)
          return losePeer("WARNING: data stream error with ", p, ev);

        consumed += msglen;
      }
      else
        break;
    }

    data.erase(data.begin(), data.begin()+consumed);

    // If the client has closed the connection, shut down our end.  If
    // we have something to send back still, leave the write direction
    // open.  Otherwise close the shop for this client.
    if (sz == 0)
      sel_.setMask(p->socket, p->mask &= ~IORead);
  }

  // Yes, please keep processing events for this socket.
  return false;
}

DQMNet& DQMNet::operator=

(

const DQMNet &

)

[private]

virtual void DQMNet::purgeDeadObjects	(	lat::Time	oldobj,
		lat::Time	deadobj
	)			[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by run().

int DQMNet::receive

(

DQMStore *

store

)

[virtual]

Reimplemented in DQMBasicNet.

Definition at line 1394 of file DQMNet.cc.

References logme().

Referenced by DQMOldReceiver::doMonitoring().

{
  logme() << "ERROR: receive() method is not supported.\n";
  return 0;
}

bool DQMNet::reconstructObject

(

Object &

o

)

[protected]

Definition at line 217 of file DQMNet.cc.

References abortReconstructObject(), DQMNet::QValue::code, extractNextObject(), label, logme(), m, lat::Regexp::match(), lat::RegexpMatch::matchString(), DQMNet::QValue::message, DQMNet::CoreObject::name, DQMNet::CoreObject::object, p, parseInt(), DQMNet::CoreObject::qreports, DQMNet::QValue::qtname, DQMNet::Object::rawdata, DQMNet::CoreObject::reference, lat::RegexpMatch::reset(), s_rxmeqr, s_rxmeval(), and value.

Referenced by reinstateObject().

{
  DQMRootBuffer buf(DQMRootBuffer::kRead, o.rawdata.size(), &o.rawdata[0], kFALSE);
  buf.Reset();

  // Extract the main object.
  if (! (o.object = extractNextObject(buf)))
    return false;
  
  // Extract the reference object.
  o.reference = extractNextObject(buf);

  // Calculate quality report base name.
  int slash = StringOps::rfind(o.name, '/');
  std::string qrbase;
  qrbase.reserve(o.name.size()+2);
  qrbase = (slash == -1 ? o.name : o.name.substr(slash+1, std::string::npos));
  qrbase += ".";

  // Extract quality reports.
  while (TObjString *qrstr = dynamic_cast<TObjString *>(extractNextObject(buf)))
  {
    RegexpMatch m;
    if (! s_rxmeval.match(qrstr->GetName(), 0, 0, &m))
    {
      logme()
        << "ERROR: unexpected quality report string '"
        << qrstr->GetName() << "' for object '"
        << o.name << "'\n";
      return abortReconstructObject(o);
    }

    std::string label = m.matchString(qrstr->GetName(), 1);
    std::string type = m.matchString(qrstr->GetName(), 2);
    std::string value = m.matchString(qrstr->GetName(), 3);

    if (type != "qr")
    {
      logme()
        << "ERROR: expected a 'qr' for a quality report '"
        << qrstr->GetName() << "' but found '" << type
        << "' instead\n";
      return abortReconstructObject(o);
    }

    std::string qrname = label;
    qrname.replace(0, qrbase.size(), "");
    if (qrname == label)
    {
      logme()
        << "ERROR: quality report label in '"
        << qrstr->GetName()
        << "' does not match object name '"
        << o.name << "'\n";
      return abortReconstructObject(o);
    }

    m.reset();
    if (! s_rxmeqr.match(value, 0, 0, &m))
    {
      logme()
        << "ERROR: quality test value '"
        << value << "' is incorrectly formatted\n";
      return abortReconstructObject(o);
    }

    QValue qval;
    qval.code = 0;
    qval.qtname = qrname;
    qval.message = m.matchString(value, 2);
    std::string strcode = m.matchString(value, 1);
    const char *p = strcode.c_str();
    if (! parseInt(p, "", 0, qval.code) || *p)
    {
      logme()
        << "ERROR: failed to determine quality test code from '"
        << value << "'\n";
      return abortReconstructObject(o);
    }

    o.qreports.push_back(qval);
  }

  return true;
}

bool DQMNet::reinstateObject	(	DQMStore *	store,
		Object &	o
	)			[protected]

Definition at line 304 of file DQMNet.cc.

References DQMStore::book1D(), DQMStore::book1S(), DQMStore::book2D(), DQMStore::book2S(), DQMStore::book3D(), DQMStore::bookFloat(), DQMStore::bookInt(), DQMStore::bookProfile(), DQMStore::bookProfile2D(), DQMStore::bookString(), e, MonitorElement::Fill(), i, logme(), m, lat::Regexp::match(), lat::RegexpMatch::matchString(), name, DQMNet::CoreObject::name, DQMNet::CoreObject::object, reconstructObject(), s_rxmeval(), DQMStore::setCurrentFolder(), t, DQMStore::tag(), DQMNet::CoreObject::tags, and value.

Referenced by DQMBasicNet::receive().

{
  if (! reconstructObject (o))
    return false;

  // Reconstruct the main object
  std::string folder = o.name;
  std::string name = o.name;
  folder.erase(folder.rfind('/'), std::string::npos);
  name.erase(0, name.rfind('/')+1);
  store->setCurrentFolder(folder);
  if (TProfile2D *t = dynamic_cast<TProfile2D *>(o.object))
    store->bookProfile2D(name, t);
  else if (TProfile *t = dynamic_cast<TProfile *>(o.object))
    store->bookProfile(name, t);
  else if (TH3F *t = dynamic_cast<TH3F *>(o.object))
    store->book3D(name, t);
  else if (TH2F *t = dynamic_cast<TH2F *>(o.object))
    store->book2D(name, t);
  else if (TH2S *t = dynamic_cast<TH2S *>(o.object))
    store->book2S(name, t);
  else if (TH1F *t = dynamic_cast<TH1F *>(o.object))
    store->book1D(name, t);
  else if (TH1S *t = dynamic_cast<TH1S *>(o.object))
    store->book1S(name, t);
  else if (TObjString *t = dynamic_cast<TObjString *>(o.object))
  {
    RegexpMatch m;
    if (! s_rxmeval.match(t->GetName(), 0, 0, &m))
    {
      logme()
        << "ERROR: unexpected monitor element string '"
        << t->GetName() << "' for object '"
        << o.name << "'\n";
      return false;
    }

    // std::string label = m.matchString(t->GetName(), 1);
    std::string type = m.matchString(t->GetName(), 2);
    std::string value = m.matchString(t->GetName(), 3);

    if (type == "i")
      store->bookInt(name)->Fill(atoi(value.c_str()));
    else if (type == "f")
      store->bookFloat(name)->Fill(atof(value.c_str()));
    else if (type == "s")
      store->bookString(name, value);
    else
    {
      logme()
        << "ERROR: unexpected string monitor element of type '"
        << type << "' (from '" << t->GetName() << "') for object '"
        << o.name << "'\n";
      return false;
    }
  }

  // Reconstruct tags.  (FIXME: untag old tags first?)
  for (size_t i = 0, e = o.tags.size(); i < e; ++i)
    store->tag(o.name, o.tags[i]);

  // FIXME: Reference and quality reports?

  // Inidicate success.
  return true;
}

void DQMNet::releaseFromWait	(	WaitList::iterator	i,
		Object *	o
	)			[private]

Definition at line 163 of file DQMNet.cc.

References alivecheck_mergeAndRegister::msg, DQMNet::Bucket::next, releaseFromWait(), and waiting_.

{
  Bucket **msg = &i->peer->sendq;
  while (*msg)
    msg = &(*msg)->next;
  *msg = new Bucket;
  (*msg)->next = 0;

  releaseFromWait(*msg, *i, o);

  assert(i->peer->waiting > 0);
  i->peer->waiting--;
  waiting_.erase(i);
}

void DQMNet::releaseFromWait	(	Bucket *	msg,
		WaitObject &	w,
		Object *	o
	)			[protected, virtual]

Definition at line 382 of file DQMNet.cc.

References copydata(), DQMNet::Bucket::data, DQM_REPLY_NONE, DQMNet::WaitObject::name, sendObjectToPeer(), and sendScalarAsText_.

Referenced by releaseFromWait(), releaseWaiters(), and run().

{
  if (o)
    sendObjectToPeer (msg, *o, true, sendScalarAsText_);
  else
  {
    uint32_t words [3];
    words[0] = sizeof(words) + w.name.size();
    words[1] = DQM_REPLY_NONE;
    words[2] = w.name.size();

    msg->data.reserve(msg->data.size() + words[0]);
    copydata(msg, &words[0], sizeof(words));
    copydata(msg, &w.name[0], w.name.size());
  }
}

void DQMNet::releaseWaiters

(

Object *

o

)

[private]

Definition at line 180 of file DQMNet.cc.

References e, i, DQMNet::CoreObject::name, releaseFromWait(), and waiting_.

Referenced by onMessage().

{
  for (WaitList::iterator i = waiting_.begin(), e = waiting_.end(); i != e; )
    if (i->name == o->name)
      releaseFromWait(i++, o);
    else
      ++i;
}

void DQMNet::removeLocalObject

(

const std::string &

name

)

[virtual]

Reimplemented in DQMBasicNet.

Definition at line 1407 of file DQMNet.cc.

References logme().

Referenced by DQMService::flush().

{
  logme() << "ERROR: removeLocalObject() method is not supported.\n";
}

virtual void DQMNet::removePeer	(	Peer *	p,
		lat::Socket *	s
	)			[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by losePeer().

void DQMNet::requestFullUpdates

(

bool

doit

)

Enable or disable requests for full updates.

Set this to get the "old" DQM networking behaviour to automatically fetch all upstream content when it changes, rather than fetching it lazily as needed. You must call this method if you use receive(); any other use is strongly discouraged. Must be called before run() or start().

Definition at line 1116 of file DQMNet.cc.

References requestFullUpdates_.

Referenced by DQMOldReceiver::DQMOldReceiver().

{
  requestFullUpdates_ = doit;
}

virtual void DQMNet::requestFullUpdatesFromPeers

(

void

)

[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, DQMBasicNet, and DQMImplNet< DQMNet::Object >.

Referenced by onMessage().

void DQMNet::requestObject	(	Peer *	p,
		const char *	name,
		size_t	len
	)			[private]

Queue an object request to the data server.

Definition at line 128 of file DQMNet.cc.

References copydata(), DQM_MSG_GET_OBJECT, alivecheck_mergeAndRegister::msg, DQMNet::Bucket::next, and DQMNet::Peer::sendq.

Referenced by onMessage(), and waitForData().

{
  Bucket **msg = &p->sendq;
  while (*msg)
    msg = &(*msg)->next;
  *msg = new Bucket;
  (*msg)->next = 0;

  uint32_t words[3];
  words[0] = sizeof(words) + len;
  words[1] = DQM_MSG_GET_OBJECT;
  words[2] = len;
  copydata(*msg, words, sizeof(words));
  copydata(*msg, name, len);
}

void DQMNet::run

(

void

)

Run the actual I/O processing loop.

Definition at line 1254 of file DQMNet.cc.

References lat::Socket::abort(), arg, lat::IOSelector::attach(), DQMNet::Peer::automatic, lat::InetSocket::connect(), copydata(), lat::CreateHook(), createPeer(), debug_, delay_, lat::IOSelector::dispatch(), downstream_, DQM_MSG_LIST_OBJECTS, DQM_MSG_UPDATE_ME, e, lat::endl(), lat::SysErr::ErrOperationInProgress, lat::Error::explain(), findObject(), flush_, DQMNet::AutoPeer::host, lat::InetAddress::hostname(), i, IORead, IOUrgent, IOWrite, lock(), logme(), DQMNet::Peer::mask, DQMNet::Bucket::next, DQMNet::AutoPeer::next, lat::Error::next(), onPeerData(), lat::SocketConst::OptSockReceiveBuffer, lat::SocketConst::OptSockSendBuffer, p, DQMNet::AutoPeer::peer, DQMNet::Peer::peeraddr, DQMNet::AutoPeer::port, lat::InetAddress::port(), lat::SystemError::portable(), purgeDeadObjects(), releaseFromWait(), requestFullUpdates_, s, sel_, sendObjectListToPeers(), DQMNet::Peer::sendq, lat::IOChannel::setBlocking(), lat::Socket::setopt(), shouldStop(), DQMNet::Peer::socket, SOCKET_BUF_SIZE, lat::SocketConst::TypeStream, unlock(), DQMNet::AutoPeer::update, DQMNet::Peer::update, updatePeerMasks(), upstream_, waiting_, and DQMNet::AutoPeer::warned.

{
  Time now;
  Time nextFlush = 0;
  AutoPeer *automatic[2] = { &upstream_, &downstream_ };

  // Perform I/O.  Every once in a while flush updates to peers.
  while (! shouldStop())
  {
    for (int i = 0; i < 2; ++i)
    {
      AutoPeer *ap = automatic[i];

      // If we need a server connection and don't have one yet,
      // initiate asynchronous connection creation.  Swallow errors
      // in case the server won't talk to us.
      if (! ap->host.empty()
          && ! ap->peer
          && (now = Time::current()) > ap->next)
      {
        ap->next = now + TimeSpan(0, 0, 0, 15 /* seconds */, 0);
        InetSocket *s = 0;
        try
        {
          s = new InetSocket (SocketConst::TypeStream);
          s->setBlocking (false);
          s->connect(InetAddress (ap->host.c_str(), ap->port));
          s->setopt(lat::SocketConst::OptSockSendBuffer, SOCKET_BUF_SIZE);
          s->setopt(lat::SocketConst::OptSockReceiveBuffer, SOCKET_BUF_SIZE);
        }
        catch (Error &e)
        {
          SystemError *sys = dynamic_cast<SystemError *>(e.next());
          if (! sys || sys->portable() != SysErr::ErrOperationInProgress)
          {
            // "In progress" just means the connection is in progress.
            // The connection is ready when the socket is writeable.
            // Anything else is a real problem.
            if (! ap->warned)
            {
              logme()
                << "NOTE: DQM server at " << ap->host << ":" << ap->port
                << " is unavailable.  Connection will be established"
                << " automatically on the background once the server"
                << " becomes available.  Error from the attempt was: "
                << e.explain() << '\n';
              ap->warned = true;
            }

            if (s)
              s->abort();
            delete s;
            s = 0;
          }
        }

        // Set up with the selector if we were successful.  If this is
        // the upstream collector, queue a request for updates.
        if (s)
        {
          lock();
          Peer *p = createPeer(s);
          ap->peer = p;
          ap->warned = false;
          unlock();

          InetAddress peeraddr = ((InetSocket *) s)->peername();
          InetAddress myaddr = ((InetSocket *) s)->sockname();
          p->peeraddr = StringFormat("%1:%2")
                        .arg(peeraddr.hostname())
                        .arg(peeraddr.port());
          p->mask = IORead|IOWrite|IOUrgent;
          p->update = ap->update;
          p->automatic = ap;
          p->socket = s;
          sel_.attach(s, p->mask, CreateHook(this, &DQMNet::onPeerData, p));
          if (ap == &upstream_)
          {
            uint32_t words[5] = { 2*sizeof(uint32_t), DQM_MSG_LIST_OBJECTS,
                                  3*sizeof(uint32_t), DQM_MSG_UPDATE_ME,
                                  requestFullUpdates_ };
            p->sendq = new Bucket;
            p->sendq->next = 0;
            copydata(p->sendq, words, sizeof(words));
          }

          // Report the new connection.
          if (debug_)
            logme()
              << "INFO: now connected to " << p->peeraddr << " from "
              << myaddr.hostname() << ":" << myaddr.port() << std::endl;
        }
      }
    }

    // Pump events for a while.
    sel_.dispatch(delay_);
    now = Time::current();

    // Check if flush is required.  Flush only if one is needed.
    // Always sends the full object list, but only rarely.
    // Compact objects no longer in active use before sending
    // out the update.
    if (flush_ && now > nextFlush)
    {
      flush_ = false;
      nextFlush = now + TimeSpan(0, 0, 0, 15 /* seconds */, 0);

      lock();
      purgeDeadObjects(now - TimeSpan(0, 0, 2 /* minutes */, 0, 0),
                       now - TimeSpan(0, 0, 20 /* minutes */, 0, 0));
      sendObjectListToPeers(true);
      unlock();
    }

    // Update the data server and peer selection masks.  If we
    // have no more data to send and listening for writes, remove
    // the write mask.  If we have something to write and aren't
    // listening for writes, start listening so we can send off
    // the data.
    updatePeerMasks();

    // Release peers that have been waiting for data for too long.
    lock();
    Time waitold = now - TimeSpan(0, 0, 2 /* minutes */, 0, 0);
    for (WaitList::iterator i = waiting_.begin(), e = waiting_.end(); i != e; )
    {
      // If the peer has waited for too long, send something.
      if (i->time < waitold)
        releaseFromWait(i++, findObject(0, i->name));

      // Keep it for now.
      else
        ++i;
    }
    unlock();
  }
}

void DQMNet::sendLocalChanges

(

void

)

Definition at line 1415 of file DQMNet.cc.

References lat::Pipe::sink(), wakeup_, and lat::IOChannel::write().

Referenced by DQMService::flush(), and DQMImplNet< DQMNet::Object >::removePeer().

{
  char byte = 0;
  wakeup_.sink()->write(&byte, 1);
}

virtual void DQMNet::sendObjectListToPeer	(	Bucket *	msg,
		bool	data,
		bool	all,
		bool	clear
	)			[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by onMessage().

virtual void DQMNet::sendObjectListToPeers

(

bool

all

)

[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by run().

void DQMNet::sendObjectToPeer	(	Bucket *	msg,
		Object &	o,
		bool	data,
		bool	text
	)			[protected]

Definition at line 432 of file DQMNet.cc.

References copydata(), DQMNet::Bucket::data, DQM_FLAG_SCALAR, DQM_FLAG_TEXT, DQM_FLAG_ZOMBIE, DQM_REPLY_OBJECT, extractScalarData(), DQMNet::CoreObject::flags, flags, DQMNet::CoreObject::name, DQMNet::Object::rawdata, DQMNet::CoreObject::tags, and DQMNet::CoreObject::version.

Referenced by onMessage(), releaseFromWait(), and DQMImplNet< DQMNet::Object >::sendObjectListToPeer().

{
  uint32_t flags = o.flags & ~DQM_FLAG_ZOMBIE;
  DataBlob objdata;

  if (text && extractScalarData(objdata, o))
    flags |= DQM_FLAG_TEXT;
  else if (data || (flags & DQM_FLAG_SCALAR))
    objdata.insert(objdata.end(),
                   &o.rawdata[0],
                   &o.rawdata[0] + o.rawdata.size());

  uint32_t words [8];
  uint32_t namelen = o.name.size();
  uint32_t taglen  = o.tags.size() * sizeof(uint32_t);
  uint32_t datalen = objdata.size();

  words[0] = 8*sizeof(uint32_t) + namelen + taglen + datalen;
  words[1] = DQM_REPLY_OBJECT;
  words[2] = flags;
  words[3] = (o.version >> 0 ) & 0xffffffff;
  words[4] = (o.version >> 32) & 0xffffffff;
  words[5] = namelen;
  words[6] = taglen / sizeof(uint32_t);
  words[7] = datalen;

  msg->data.reserve(msg->data.size() + words[0]);
  copydata(msg, &words[0], 8*sizeof(uint32_t));
  if (namelen)
    copydata(msg, &o.name[0], namelen);
  if (taglen)
    copydata(msg, &o.tags[0], taglen);
  if (datalen)
    copydata(msg, &objdata[0], datalen);
}

void DQMNet::sendScalarAsText

(

bool

doit

)

Enable or disable sending scalar monitoring values as text, rather than their ROOT object values.

Must be called before run() or start().

Definition at line 1105 of file DQMNet.cc.

References sendScalarAsText_.

{
  sendScalarAsText_ = doit;
}

bool DQMNet::shouldStop

(

void

)

[protected, virtual]

Definition at line 374 of file DQMNet.cc.

References shutdown_.

Referenced by run().

{
  return shutdown_;
}

void DQMNet::shutdown

(

void

)

Stop the network layer and wait it to finish.

Definition at line 1198 of file DQMNet.cc.

References communicate_, and shutdown_.

Referenced by DQMService::shutdown().

{
  shutdown_ = 1;
  if (communicate_ != (pthread_t) -1)
    pthread_join(communicate_, 0);
}

void DQMNet::start

(

void

)

Start running the network layer in a new thread.

This is an exclusive alternative to the run() method, which runs the network layer in the caller's thread.

Definition at line 1239 of file DQMNet.cc.

References communicate(), communicate_, lock_, logme(), and pthread_create.

Referenced by DQMService::DQMService().

{
  if (communicate_ != (pthread_t) -1)
  {
    logme()
      << "ERROR: DQM networking thread has already been started\n";
    return;
  }

  pthread_mutex_init(&lock_, 0);
  pthread_create (&communicate_, 0, &communicate, this);
}

void DQMNet::startLocalServer

(

int

port

)

Start a server socket for accessing this DQM node remotely.

Must be called before calling run() or start(). May throw an Exception if the server socket cannot be initialised.

Definition at line 1125 of file DQMNet.cc.

References lat::IOSelector::attach(), lat::CreateHook(), e, lat::endl(), Exception, lat::Error::explain(), IOAccept, logme(), onPeerConnect(), lat::SocketConst::OptSockReceiveBuffer, lat::SocketConst::OptSockSendBuffer, sel_, server_, lat::IOChannel::setBlocking(), lat::Socket::setopt(), and SOCKET_BUF_SIZE.

{
  if (server_)
  {
    logme() << "ERROR: DQM server was already started.\n";
    return;
  }

  try
  {
    server_ = new InetServerSocket(InetAddress (port), 10);
    server_->setopt(lat::SocketConst::OptSockSendBuffer, SOCKET_BUF_SIZE);
    server_->setopt(lat::SocketConst::OptSockReceiveBuffer, SOCKET_BUF_SIZE);
    server_->setBlocking(false);
    sel_.attach(server_, IOAccept, CreateHook(this, &DQMNet::onPeerConnect));
  }
  catch (Error &e)
  {
    // FIXME: Do we need to do this when we throw an exception anyway?
    // FIXME: Abort instead?
    logme()
      << "ERROR: Failed to start server at port " << port << ": "
      << e.explain() << std::endl;

    // FIXME: Throw something simpler that removes the dependency?
    throw cms::Exception("DQMNet::startLocalServer")
      << "Failed to start server at port " << port << ": "
      << e.explain();
  }
  
  logme() << "INFO: DQM server started at port " << port << std::endl;
}

void DQMNet::unlock

(

void

)

Release the lock on the DQM net layer.

Definition at line 1229 of file DQMNet.cc.

References communicate_, and lock_.

Referenced by DQMService::flush(), onMessage(), DQMBasicNet::receive(), and run().

{
  if (communicate_ != (pthread_t) -1)
    pthread_mutex_unlock(&lock_);
}

void DQMNet::updateLocalObject

(

Object &

o

)

[virtual]

Reimplemented in DQMBasicNet.

Definition at line 1401 of file DQMNet.cc.

References logme().

Referenced by DQMService::flush().

{
  logme() << "ERROR: updateLocalObject() method is not supported.\n";
}

void DQMNet::updateMask

(

Peer *

p

)

[protected]

Update the selector mask for a peer based on data queues.

Close the connection if there is no reason to maintain it open.

Definition at line 1023 of file DQMNet.cc.

References debug_, lat::endl(), IOUrgent, IOWrite, logme(), losePeer(), DQMNet::Peer::mask, DQMNet::Peer::peeraddr, sel_, DQMNet::Peer::sendq, lat::IOSelector::setMask(), DQMNet::Peer::socket, and DQMNet::Peer::waiting.

Referenced by DQMImplNet< DQMNet::Object >::updatePeerMasks().

{
  if (! p->socket)
    return;

  // Listen to writes iff we have data to send.
  unsigned oldmask = p->mask;
  if (! p->sendq && (p->mask & IOWrite))
    sel_.setMask(p->socket, p->mask &= ~IOWrite);

  if (p->sendq && ! (p->mask & IOWrite))
    sel_.setMask(p->socket, p->mask |= IOWrite);

  if (debug_ && oldmask != p->mask)
    logme()
      << "DEBUG: updating mask for " << p->peeraddr << " to "
      << p->mask << " from " << oldmask << std::endl;

  // If we have nothing more to send and are no longer listening
  // for reads, close up the shop for this peer.
  if (p->mask == IOUrgent && ! p->waiting)
  {
    assert (! p->sendq);
    if (debug_)
      logme() << "INFO: connection closed to " << p->peeraddr << std::endl;
    losePeer(0, p, 0);
  }
}

virtual void DQMNet::updatePeerMasks

(

void

)

[protected, pure virtual]

Implemented in DQMImplNet< ObjType >, and DQMImplNet< DQMNet::Object >.

Referenced by run().

void DQMNet::updateToCollector	(	const std::string &	host,
		int	port
	)

Tell the network layer to connect to host and port and automatically send updates whenever local DQM data changes.

Must be called before calling run() or start().

Definition at line 1162 of file DQMNet.cc.

References downstream_, lat::endl(), DQMNet::AutoPeer::host, logme(), DQMNet::AutoPeer::port, and DQMNet::AutoPeer::update.

Referenced by DQMService::DQMService().

{
  if (! downstream_.host.empty())
  {
    logme()
      << "ERROR: Already updating another collector at "
      << downstream_.host << ":" << downstream_.port << std::endl;
    return;
  }

  downstream_.update = true;
  downstream_.host = host;
  downstream_.port = port;
}

void DQMNet::waitForData	(	Peer *	p,
		const std::string &	name,
		const std::string &	info,
		Peer *	owner
	)			[protected]

Queue a request for an object and put a peer into the mode of waiting for object data to appear.

Definition at line 147 of file DQMNet.cc.

References requestObject(), DQMNet::Peer::waiting, and waiting_.

Referenced by onMessage().

{
  // FIXME: Should we automatically record which exact peer the waiter
  // is expecting to deliver data so we know to release the waiter if
  // the other peer vanishes?  The current implementation stands a
  // chance for the waiter to wait indefinitely -- although we do
  // force terminate the wait after a while.
  requestObject(owner, name.size() ? &name[0] : 0, name.size());
  WaitObject wo = { Time::current(), name, info, p };
  waiting_.push_back(wo);
  p->waiting++;
}

---

Member Data Documentation

std::string DQMNet::appname_ [private]

Definition at line 193 of file DQMNet.h.

Referenced by logme().

pthread_t DQMNet::communicate_ [private]

Definition at line 206 of file DQMNet.h.

Referenced by lock(), shutdown(), start(), and unlock().

bool DQMNet::debug_ [protected]

Definition at line 175 of file DQMNet.h.

Referenced by debug(), onMessage(), onPeerConnect(), onPeerData(), DQMImplNet< DQMNet::Object >::purgeDeadObjects(), run(), DQMImplNet< DQMNet::Object >::sendObjectListToPeers(), and updateMask().

int DQMNet::delay_ [private]

Definition at line 209 of file DQMNet.h.

Referenced by delay(), and run().

AutoPeer DQMNet::downstream_ [private]

Definition at line 202 of file DQMNet.h.

Referenced by DQMNet(), run(), and updateToCollector().

const uint32_t DQMNet::DQM_FLAG_DEAD = 0x80000000 [static]

Definition at line 42 of file DQMNet.h.

Referenced by DQMImplNet< DQMNet::Object >::markObjectsDead(), onMessage(), DQMImplNet< DQMNet::Object >::purgeDeadObjects(), and DQMBasicNet::receive().

const uint32_t DQMNet::DQM_FLAG_NEW = 0x40000000 [static]

Definition at line 41 of file DQMNet.h.

Referenced by MonitorElement::MonitorElement(), onMessage(), DQMImplNet< DQMNet::Object >::sendObjectListToPeer(), MonitorElement::update(), and MonitorElement::wasUpdated().

const uint32_t DQMNet::DQM_FLAG_RECEIVED = 0x20000000 [static]

Definition at line 40 of file DQMNet.h.

Referenced by onMessage(), and DQMBasicNet::receive().

const uint32_t DQMNet::DQM_FLAG_REPORT_ERROR = 0x1 [static]

Definition at line 34 of file DQMNet.h.

Referenced by MonitorElement::updateQReportStats().

const uint32_t DQMNet::DQM_FLAG_REPORT_OTHER = 0x4 [static]

Definition at line 36 of file DQMNet.h.

Referenced by MonitorElement::updateQReportStats().

const uint32_t DQMNet::DQM_FLAG_REPORT_WARNING = 0x2 [static]

Definition at line 35 of file DQMNet.h.

Referenced by MonitorElement::updateQReportStats().

const uint32_t DQMNet::DQM_FLAG_SCALAR = 0x8 [static]

Definition at line 37 of file DQMNet.h.

Referenced by extractScalarData(), MonitorElement::initialise(), DQMImplNet< DQMNet::Object >::purgeDeadObjects(), and sendObjectToPeer().

const uint32_t DQMNet::DQM_FLAG_TEXT = 0x10000000 [static]

Definition at line 39 of file DQMNet.h.

Referenced by sendObjectToPeer().

const uint32_t DQMNet::DQM_FLAG_ZOMBIE = 0x08000000 [static]

Definition at line 38 of file DQMNet.h.

Referenced by DQMImplNet< DQMNet::Object >::markObjectsDead(), DQMImplNet< DQMNet::Object >::markObjectsZombies(), and sendObjectToPeer().

const uint32_t DQMNet::DQM_MSG_GET_OBJECT = 3 [static]

Definition at line 27 of file DQMNet.h.

Referenced by onMessage(), and requestObject().

const uint32_t DQMNet::DQM_MSG_HELLO = 0 [static]

Definition at line 24 of file DQMNet.h.

const uint32_t DQMNet::DQM_MSG_LIST_OBJECTS = 2 [static]

Definition at line 26 of file DQMNet.h.

Referenced by onMessage(), and run().

const uint32_t DQMNet::DQM_MSG_UPDATE_ME = 1 [static]

Definition at line 25 of file DQMNet.h.

Referenced by onMessage(), DQMBasicNet::requestFullUpdatesFromPeers(), and run().

const uint32_t DQMNet::DQM_REPLY_LIST_BEGIN = 101 [static]

Definition at line 29 of file DQMNet.h.

Referenced by onMessage(), and DQMImplNet< DQMNet::Object >::sendObjectListToPeer().

const uint32_t DQMNet::DQM_REPLY_LIST_END = 102 [static]

Definition at line 30 of file DQMNet.h.

Referenced by onMessage(), and DQMImplNet< DQMNet::Object >::sendObjectListToPeer().

const uint32_t DQMNet::DQM_REPLY_NONE = 103 [static]

Definition at line 31 of file DQMNet.h.

Referenced by onMessage(), and releaseFromWait().

const uint32_t DQMNet::DQM_REPLY_OBJECT = 104 [static]

Definition at line 32 of file DQMNet.h.

Referenced by onMessage(), and sendObjectToPeer().

bool DQMNet::flush_ [private]

Definition at line 210 of file DQMNet.h.

Referenced by onLocalNotify(), onMessage(), and run().

pthread_mutex_t DQMNet::lock_ [private]

Definition at line 205 of file DQMNet.h.

Referenced by lock(), start(), and unlock().

const uint32_t DQMNet::MAX_PEER_WAITREQS = 128 [static]

Definition at line 44 of file DQMNet.h.

Referenced by onPeerData().

int DQMNet::pid_ [private]

Definition at line 194 of file DQMNet.h.

Referenced by logme().

bool DQMNet::requestFullUpdates_ [protected]

Definition at line 177 of file DQMNet.h.

Referenced by onMessage(), requestFullUpdates(), and run().

lat::IOSelector DQMNet::sel_ [private]

Definition at line 196 of file DQMNet.h.

Referenced by DQMNet(), losePeer(), onPeerConnect(), onPeerData(), run(), startLocalServer(), and updateMask().

bool DQMNet::sendScalarAsText_ [protected]

Definition at line 176 of file DQMNet.h.

Referenced by onMessage(), releaseFromWait(), DQMImplNet< DQMNet::Object >::sendObjectListToPeer(), and sendScalarAsText().

lat::InetServerSocket* DQMNet::server_ [private]

Definition at line 197 of file DQMNet.h.

Referenced by onPeerConnect(), and startLocalServer().

sig_atomic_t DQMNet::shutdown_ [private]

Definition at line 207 of file DQMNet.h.

Referenced by shouldStop(), and shutdown().

AutoPeer DQMNet::upstream_ [private]

Definition at line 201 of file DQMNet.h.

Referenced by DQMNet(), listenToCollector(), and run().

lat::Time DQMNet::version_ [private]

Definition at line 199 of file DQMNet.h.

WaitList DQMNet::waiting_ [private]

Definition at line 203 of file DQMNet.h.

Referenced by losePeer(), releaseFromWait(), releaseWaiters(), run(), and waitForData().

lat::Pipe DQMNet::wakeup_ [private]

Definition at line 198 of file DQMNet.h.

Referenced by DQMNet(), and sendLocalChanges().

---

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

• DQMServices/Core/interface/DQMNet.h
• DQMServices/Core/src/DQMNet.cc

---