DQMImplNet< ObjType > Class Template Reference

#include <DQMNet.h>

Inheritance diagram for DQMImplNet< ObjType >:

Classes
struct	ImplPeer

Public Types
using	DirMap = std::set< std::string >
typedef std::unordered_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	QValue = MonitorElementData::QReport::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_TH1I = 0x00000013
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_TH2I = 0x00000023
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 381 of file DQMNet.h.

Member Typedef Documentation

DirMap

template<class ObjType>

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

Definition at line 385 of file DQMNet.h.

ObjectMap

template<class ObjType>

typedef std::unordered_set<ObjType, HashOp, HashEqual> DQMImplNet< ObjType >::ObjectMap

Definition at line 386 of file DQMNet.h.

PeerMap

template<class ObjType>

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

Definition at line 387 of file DQMNet.h.

Constructor & Destructor Documentation

DQMImplNet()

template<class ObjType>

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

inline

Definition at line 394 of file DQMNet.h.

: DQMNet(appname) {}

~DQMImplNet()

template<class ObjType>

DQMImplNet< ObjType >::~DQMImplNet ( )

overridedefault

Member Function Documentation

createPeer()

template<class ObjType>

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

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 488 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;
  }

findObject()

template<class ObjType>

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

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 399 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;
    }
  }

getPeer()

template<class ObjType>

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

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 482 of file DQMNet.h.

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

makeObject()

template<class ObjType>

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

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 436 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);
  }

markObjectsDead()

template<class ObjType>

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

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 459 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;
    }
  }

purgeDeadObjects()

template<class ObjType>

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

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 471 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;
    }
  }

removePeer()

template<class ObjType>

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

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 503 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();
  }

sendObjectListToPeer()

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 520 of file DQMNet.h.

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

                                                                        {
    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));
  }

sendObjectListToPeers()

template<class ObjType>

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

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 555 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;
    }
  }

updatePeerMasks()

template<class ObjType>

void DQMImplNet< ObjType >::updatePeerMasks ( )

inlineoverrideprotectedvirtual

Implements DQMNet.

Definition at line 583 of file DQMNet.h.

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

Member Data Documentation

peers_

template<class ObjType>

PeerMap DQMImplNet< ObjType >::peers_

protected

Definition at line 590 of file DQMNet.h.

Referenced by DQMImplNet< DQMNet::Object >::createPeer(), DQMImplNet< DQMNet::Object >::findObject(), DQMImplNet< DQMNet::Object >::getPeer(), DQMImplNet< DQMNet::Object >::removePeer(), DQMImplNet< DQMNet::Object >::sendObjectListToPeer(), DQMImplNet< DQMNet::Object >::sendObjectListToPeers(), and DQMImplNet< DQMNet::Object >::updatePeerMasks().