#include <DQMNet.h>

Inheritance diagram for DQMImplNet< ObjType >:

Classes
struct	ImplPeer

Public Types
using	DirMap = std::set< std::string >

typedef __gnu_cxx::hash_set< ObjType, HashOp, HashEqual >	ObjectMap

typedef std::map< lat::Socket *, ImplPeer >	PeerMap

Public Types inherited from DQMNet
using	DataBlob = std::vector< unsigned char >

using	QReports = std::vector< QValue >

using	TagList = std::vector< uint32_t >

using	WaitList = std::list< WaitObject >

Public Member Functions
	DQMImplNet (const std::string &appname="")

	~DQMImplNet () override=default

Public Member Functions inherited from DQMNet
void	debug (bool doit)

void	delay (int delay)

	DQMNet (const std::string &appname="")

	DQMNet (const DQMNet &)=delete

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

void	lock ()
	Acquire a lock on the DQM net layer. More...

DQMNet &	operator= (const DQMNet &)=delete

void	run ()

void	sendLocalChanges ()

void	shutdown ()
	Stop the network layer and wait it to finish. More...

void	staleObjectWaitLimit (lat::TimeSpan time)

void	start ()

void	startLocalServer (int port)

void	startLocalServer (const char *path)

void	unlock ()
	Release the lock on the DQM net layer. More...

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

virtual	~DQMNet ()

Protected Member Functions
Peer *	createPeer (lat::Socket *s) override

Object *	findObject (Peer p, const std::string &name, Peer *owner=nullptr) override

Peer *	getPeer (lat::Socket *s) override

Object *	makeObject (Peer *p, const std::string &name) override

void	markObjectsDead (Peer *p) override

void	purgeDeadObjects (Peer *p) override

void	removePeer (Peer p, lat::Socket s) override

void	sendObjectListToPeer (Bucket *msg, bool all, bool clear) override
	Send all objects to a peer and optionally mark sent objects old. More...

void	sendObjectListToPeers (bool all) override

void	updatePeerMasks () override

Protected Member Functions inherited from DQMNet
std::ostream &	logme ()

virtual bool	onMessage (Bucket msg, Peer p, unsigned char *data, size_t len)

virtual void	releaseFromWait (Bucket msg, WaitObject &w, Object o)

virtual void	sendObjectToPeer (Bucket *msg, Object &o, bool data)

virtual bool	shouldStop ()

void	updateMask (Peer *p)

void	waitForData (Peer p, const std::string &name, const std::string &info, Peer owner)

Protected Attributes
PeerMap	peers_

Protected Attributes inherited from DQMNet
bool	debug_

pthread_mutex_t	lock_

Additional Inherited Members
Static Public Member Functions inherited from DQMNet
static size_t	dqmhash (const void *key, size_t keylen)

static void	packQualityData (std::string &into, const QReports &qr)

static bool	setOrder (const CoreObject &a, const CoreObject &b)

static void	unpackQualityData (QReports &qr, uint32_t &flags, const char *from)

Static Public Attributes inherited from DQMNet
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_PROP_ACCUMULATE = 0x00004000

static const uint32_t	DQM_PROP_DEAD = 0x00080000

static const uint32_t	DQM_PROP_EFFICIENCY_PLOT = 0x00200000

static const uint32_t	DQM_PROP_HAS_REFERENCE = 0x00001000

static const uint32_t	DQM_PROP_LUMI = 0x00040000

static const uint32_t	DQM_PROP_MARKTODELETE = 0x01000000

static const uint32_t	DQM_PROP_NEW = 0x00010000

static const uint32_t	DQM_PROP_RECEIVED = 0x00020000

static const uint32_t	DQM_PROP_REPORT_ALARM = (DQM_PROP_REPORT_ERROR \| DQM_PROP_REPORT_WARN \| DQM_PROP_REPORT_OTHER)

static const uint32_t	DQM_PROP_REPORT_CLEAR = 0x00000000

static const uint32_t	DQM_PROP_REPORT_ERROR = 0x00000100

static const uint32_t	DQM_PROP_REPORT_MASK = 0x00000f00

static const uint32_t	DQM_PROP_REPORT_OTHER = 0x00000400

static const uint32_t	DQM_PROP_REPORT_WARN = 0x00000200

static const uint32_t	DQM_PROP_RESET = 0x00008000

static const uint32_t	DQM_PROP_STALE = 0x00100000

static const uint32_t	DQM_PROP_TAGGED = 0x00002000

static const uint32_t	DQM_PROP_TYPE_DATABLOB = 0x00000050

static const uint32_t	DQM_PROP_TYPE_INT = 0x00000001

static const uint32_t	DQM_PROP_TYPE_INVALID = 0x00000000

static const uint32_t	DQM_PROP_TYPE_MASK = 0x000000ff

static const uint32_t	DQM_PROP_TYPE_REAL = 0x00000002

static const uint32_t	DQM_PROP_TYPE_SCALAR = 0x0000000f

static const uint32_t	DQM_PROP_TYPE_STRING = 0x00000003

static const uint32_t	DQM_PROP_TYPE_TH1D = 0x00000012

static const uint32_t	DQM_PROP_TYPE_TH1F = 0x00000010

static const uint32_t	DQM_PROP_TYPE_TH1S = 0x00000011

static const uint32_t	DQM_PROP_TYPE_TH2D = 0x00000022

static const uint32_t	DQM_PROP_TYPE_TH2F = 0x00000020

static const uint32_t	DQM_PROP_TYPE_TH2S = 0x00000021

static const uint32_t	DQM_PROP_TYPE_TH3D = 0x00000032

static const uint32_t	DQM_PROP_TYPE_TH3F = 0x00000030

static const uint32_t	DQM_PROP_TYPE_TH3S = 0x00000031

static const uint32_t	DQM_PROP_TYPE_TPROF = 0x00000040

static const uint32_t	DQM_PROP_TYPE_TPROF2D = 0x00000041

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

Static Protected Member Functions inherited from DQMNet
static void	copydata (Bucket b, const void data, size_t len)

static void	discard (Bucket *&b)

Detailed Description

template<class ObjType>
class DQMImplNet< ObjType >

Definition at line 384 of file DQMNet.h.

Member Typedef Documentation

template<class ObjType>

using DQMImplNet< ObjType >::DirMap = std::set<std::string>

Definition at line 388 of file DQMNet.h.

template<class ObjType>

typedef __gnu_cxx::hash_set<ObjType, HashOp, HashEqual> DQMImplNet< ObjType >::ObjectMap

Definition at line 389 of file DQMNet.h.

template<class ObjType>

typedef std::map<lat::Socket *, ImplPeer> DQMImplNet< ObjType >::PeerMap

Definition at line 390 of file DQMNet.h.

Constructor & Destructor Documentation

template<class ObjType>

DQMImplNet< ObjType >::DQMImplNet ( const std::string & appname = "" )

inline

Definition at line 397 of file DQMNet.h.

397 : DQMNet(appname) {}

DQMNet::DQMNet

DQMNet(const std::string &appname="")

Definition: DQMNet.cc:897

template<class ObjType>

DQMImplNet< ObjType >::~DQMImplNet ( )

overridedefault

Member Function Documentation

template<class ObjType>

Peer* DQMImplNet< ObjType >::createPeer ( lat::Socket * s )

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 491 of file DQMNet.h.

                                           {
     ImplPeer *ip = &peers_[s];
     ip->socket = nullptr;
     ip->sendq = nullptr;
     ip->sendpos = 0;
     ip->mask = 0;
     ip->source = false;
     ip->update = false;
     ip->updated = false;
     ip->updates = 0;
     ip->waiting = 0;
     ip->automatic = nullptr;
     return ip;
   }

template<class ObjType>

Object* DQMImplNet< ObjType >::findObject	(	Peer *	p,
		const std::string &	name,
		Peer **	owner = `nullptr`
	)

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 402 of file DQMNet.h.

                                                                                       {
     size_t slash = name.rfind('/');
     size_t dirpos = (slash == std::string::npos ? 0 : slash);
     size_t namepos = (slash == std::string::npos ? 0 : slash + 1);
     std::string path(name, 0, dirpos);
     ObjType proto;
     proto.hash = dqmhash(name.c_str(), name.size());
     proto.dirname = &path;
     proto.objname.append(name, namepos, std::string::npos);
 
     typename ObjectMap::iterator pos;
     typename PeerMap::iterator i, e;
     if (owner)
       *owner = nullptr;
     if (p) {
       auto *ip = static_cast<ImplPeer *>(p);
       pos = ip->objs.find(proto);
       if (pos == ip->objs.end())
         return nullptr;
       else {
         if (owner)
           *owner = ip;
         return const_cast<ObjType *>(&*pos);
       }
     } else {
       for (i = peers_.begin(), e = peers_.end(); i != e; ++i) {
         pos = i->second.objs.find(proto);
         if (pos != i->second.objs.end()) {
           if (owner)
             *owner = &i->second;
           return const_cast<ObjType *>(&*pos);
         }
       }
       return nullptr;
     }
   }

template<class ObjType>

Peer* DQMImplNet< ObjType >::getPeer ( lat::Socket * s )

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 485 of file DQMNet.h.

                                        {
     auto pos = peers_.find(s);
     auto end = peers_.end();
     return pos == end ? nullptr : &pos->second;
   }

template<class ObjType>

Object* DQMImplNet< ObjType >::makeObject	(	Peer *	p,
		const std::string &	name
	)

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 439 of file DQMNet.h.

                                                               {
     auto *ip = static_cast<ImplPeer *>(p);
     size_t slash = name.rfind('/');
     size_t dirpos = (slash == std::string::npos ? 0 : slash);
     size_t namepos = (slash == std::string::npos ? 0 : slash + 1);
     ObjType o;
     o.flags = 0;
     o.tag = 0;
     o.version = 0;
     o.lastreq = 0;
     o.dirname = &*ip->dirs.insert(name.substr(0, dirpos)).first;
     o.objname.append(name, namepos, std::string::npos);
     o.hash = dqmhash(name.c_str(), name.size());
     return const_cast<ObjType *>(&*ip->objs.insert(o).first);
   }

template<class ObjType>

void DQMImplNet< ObjType >::markObjectsDead ( Peer * p )

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 462 of file DQMNet.h.

                                          {
     uint64_t minreq = (lat::Time::current() - lat::TimeSpan(0, 0, 5 /* minutes */, 0, 0)).ns();
     auto *ip = static_cast<ImplPeer *>(p);
     typename ObjectMap::iterator i, e;
     for (i = ip->objs.begin(), e = ip->objs.end(); i != e; ++i) {
       if (i->lastreq && i->lastreq < minreq)
         const_cast<ObjType &>(*i).lastreq = 0;
       const_cast<ObjType &>(*i).flags |= DQM_PROP_DEAD;
     }
   }

template<class ObjType>

void DQMImplNet< ObjType >::purgeDeadObjects ( Peer * p )

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 474 of file DQMNet.h.

                                           {
     auto *ip = static_cast<ImplPeer *>(p);
     typename ObjectMap::iterator i, e;
     for (i = ip->objs.begin(), e = ip->objs.end(); i != e;) {
       if (i->flags & DQM_PROP_DEAD)
         ip->objs.erase(i++);
       else
         ++i;
     }
   }

template<class ObjType>

void DQMImplNet< ObjType >::removePeer	(	Peer *	p,
		lat::Socket *	s
	)

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 506 of file DQMNet.h.

                                                   {
     auto *ip = static_cast<ImplPeer *>(p);
     bool needflush = !ip->objs.empty();
 
     typename ObjectMap::iterator i, e;
     for (i = ip->objs.begin(), e = ip->objs.end(); i != e;)
       ip->objs.erase(i++);
 
     peers_.erase(s);
 
     // If we removed a peer with objects, our list of objects
     // has changed and we need to update downstream peers.
     if (needflush)
       sendLocalChanges();
   }

template<class ObjType>

void DQMImplNet< ObjType >::sendObjectListToPeer	(	Bucket *	msg,
		bool	all,
		bool	clear
	)

inlineoverrideprotectedvirtual

Send all objects to a peer and optionally mark sent objects old.

Implements DQMNet.

Definition at line 523 of file DQMNet.h.

                                                                         {
     typename PeerMap::iterator pi, pe;
     typename ObjectMap::iterator oi, oe;
     size_t size = 0;
     size_t numobjs = 0;
     for (pi = peers_.begin(), pe = peers_.end(); pi != pe; ++pi)
       for (oi = pi->second.objs.begin(), oe = pi->second.objs.end(); oi != oe; ++oi, ++numobjs)
         if (all || (oi->flags & DQM_PROP_NEW))
           size += 9 * sizeof(uint32_t) + oi->dirname->size() + oi->objname.size() + 1 + oi->scalar.size() +
                   oi->qdata.size() + (oi->lastreq > 0 ? oi->rawdata.size() : 0);
 
     msg->data.reserve(msg->data.size() + size + 8 * sizeof(uint32_t));
 
     uint32_t nupdates = 0;
     uint32_t words[4];
     words[0] = sizeof(words);
     words[1] = DQM_REPLY_LIST_BEGIN;
     words[2] = numobjs;
     words[3] = all;
     copydata(msg, &words[0], sizeof(words));
 
     for (pi = peers_.begin(), pe = peers_.end(); pi != pe; ++pi)
       for (oi = pi->second.objs.begin(), oe = pi->second.objs.end(); oi != oe; ++oi)
         if (all || (oi->flags & DQM_PROP_NEW)) {
           sendObjectToPeer(msg, const_cast<ObjType &>(*oi), oi->lastreq > 0);
           if (clear)
             const_cast<ObjType &>(*oi).flags &= ~DQM_PROP_NEW;
           ++nupdates;
         }
 
     words[1] = DQM_REPLY_LIST_END;
     words[2] = nupdates;
     copydata(msg, &words[0], sizeof(words));
   }

template<class ObjType>

void DQMImplNet< ObjType >::sendObjectListToPeers ( bool all )

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 558 of file DQMNet.h.

                                                 {
     typename PeerMap::iterator i, e;
     typename ObjectMap::iterator oi, oe;
     for (i = peers_.begin(), e = peers_.end(); i != e; ++i) {
       ImplPeer &p = i->second;
       if (!p.update)
         continue;
 
       if (debug_)
         logme() << "DEBUG: notifying " << p.peeraddr << std::endl;
 
       Bucket msg;
       msg.next = nullptr;
       sendObjectListToPeer(&msg, !p.updated || all, true);
 
       if (!msg.data.empty()) {
         Bucket **prev = &p.sendq;
         while (*prev)
           prev = &(*prev)->next;
 
         *prev = new Bucket;
         (*prev)->next = nullptr;
         (*prev)->data.swap(msg.data);
       }
       p.updated = true;
     }
   }

template<class ObjType>

void DQMImplNet< ObjType >::updatePeerMasks ( )

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 586 of file DQMNet.h.

                                   {
     typename PeerMap::iterator i, e;
     for (i = peers_.begin(), e = peers_.end(); i != e;)
       updateMask(&(i++)->second);
   }

Member Data Documentation

template<class ObjType>

PeerMap DQMImplNet< ObjType >::peers_

protected

Definition at line 593 of file DQMNet.h.

Classes

Public Types

Public Member Functions

Protected Member Functions

Protected Attributes

Additional Inherited Members

Detailed Description

template<class ObjType> class DQMImplNet< ObjType >

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation

template<class ObjType>
class DQMImplNet< ObjType >