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

Member Typedef Documentation

template<class ObjType>

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

Definition at line 366 of file DQMNet.h.

template<class ObjType>

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

Definition at line 367 of file DQMNet.h.

template<class ObjType>

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

Definition at line 368 of file DQMNet.h.

Constructor & Destructor Documentation

template<class ObjType>

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

inline

Definition at line 376 of file DQMNet.h.

    : DQMNet(appname)
    {}

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 492 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 385 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 484 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 428 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 453 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 470 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 509 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 528 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 567 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 600 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 608 of file DQMNet.h.