DQMNet Class Reference

#include <DQMNet.h>

Inheritance diagram for DQMNet:

Classes
struct	AutoPeer
struct	Bucket
struct	CoreObject
struct	HashEqual
struct	HashOp
struct	Object
struct	Peer
struct	WaitObject

Public Types
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
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 ()

Static Public Member Functions
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
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

Protected Member Functions
virtual Peer *	createPeer (lat::Socket *s)=0
virtual Object *	findObject (Peer *p, const std::string &name, Peer **owner=nullptr)=0
virtual Peer *	getPeer (lat::Socket *s)=0
std::ostream &	logme ()
virtual Object *	makeObject (Peer *p, const std::string &name)=0
virtual void	markObjectsDead (Peer *p)=0
virtual bool	onMessage (Bucket *msg, Peer *p, unsigned char *data, size_t len)
virtual void	purgeDeadObjects (Peer *p)=0
virtual void	releaseFromWait (Bucket *msg, WaitObject &w, Object *o)
virtual void	removePeer (Peer *p, lat::Socket *s)=0
virtual void	sendObjectListToPeer (Bucket *msg, bool all, bool clear)=0
virtual void	sendObjectListToPeers (bool all)=0
virtual void	sendObjectToPeer (Bucket *msg, Object &o, bool data)
virtual bool	shouldStop ()
void	updateMask (Peer *p)
virtual void	updatePeerMasks ()=0
void	waitForData (Peer *p, const std::string &name, const std::string &info, Peer *owner)

Static Protected Member Functions
static void	copydata (Bucket *b, const void *data, size_t len)
static void	discard (Bucket *&b)

Protected Attributes
bool	debug_
pthread_mutex_t	lock_

Private Member Functions
void	losePeer (const char *reason, Peer *peer, lat::IOSelectEvent *event, lat::Error *err=nullptr)
bool	onLocalNotify (lat::IOSelectEvent *ev)
bool	onPeerConnect (lat::IOSelectEvent *ev)
bool	onPeerData (lat::IOSelectEvent *ev, Peer *p)
	Handle communication to a particular client. More...
void	releaseFromWait (WaitList::iterator i, Object *o)
void	releaseWaiters (const std::string &name, Object *o)
void	requestObjectData (Peer *p, const char *name, size_t len)
	Queue an object request to the data server. More...

Private Attributes
std::string	appname_
pthread_t	communicate_
int	delay_
AutoPeer	downstream_
bool	flush_
int	pid_
lat::IOSelector	sel_
lat::Socket *	server_
sig_atomic_t	shutdown_
AutoPeer	upstream_
lat::Time	version_
WaitList	waiting_
lat::TimeSpan	waitMax_
lat::TimeSpan	waitStale_
lat::Pipe	wakeup_

Detailed Description

Definition at line 26 of file DQMNet.h.

Member Typedef Documentation

DataBlob

using DQMNet::DataBlob = std::vector<unsigned char>

Definition at line 85 of file DQMNet.h.

QReports

using DQMNet::QReports = std::vector<QValue>

Definition at line 86 of file DQMNet.h.

QValue

using DQMNet::QValue = MonitorElementData::QReport::QValue

Definition at line 84 of file DQMNet.h.

TagList

using DQMNet::TagList = std::vector<uint32_t>

Definition at line 87 of file DQMNet.h.

WaitList

using DQMNet::WaitList = std::list<WaitObject>

Definition at line 88 of file DQMNet.h.

Constructor & Destructor Documentation

DQMNet() [1/2]

DQMNet::DQMNet

(

const std::string &

appname = ""

)

Definition at line 914 of file DQMNet.cc.

References downstream_, edm::storage::IORead, DQMNet::AutoPeer::next, O_NONBLOCK, onLocalNotify(), DQMNet::AutoPeer::peer, DQMNet::AutoPeer::port, sel_, DQMNet::AutoPeer::update, upstream_, and wakeup_.

    : debug_(false),
      appname_(appname.empty() ? "DQMNet" : appname.c_str()),
      pid_(getpid()),
      server_(nullptr),
      version_(Time::current()),
      communicate_((pthread_t)-1),
      shutdown_(0),
      delay_(1000),
      waitStale_(0, 0, 0, 0, 500000000 /* 500 ms */),
      waitMax_(0, 0, 0, 5 /* seconds */, 0),
      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 = nullptr;
  upstream_.next = downstream_.next = 0;
  upstream_.port = downstream_.port = 0;
  upstream_.update = downstream_.update = false;
}

~DQMNet()

DQMNet::~DQMNet ( )

virtual

Definition at line 939 of file DQMNet.cc.

                {
  // FIXME
}

DQMNet() [2/2]

DQMNet::DQMNet ( const DQMNet &  )

delete

Member Function Documentation

copydata()

void DQMNet::copydata ( Bucket *  b, const void *  data, size_t  len  )

staticprotected

Definition at line 57 of file DQMNet.cc.

References b, and data.

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

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

createPeer()

virtual Peer* DQMNet::createPeer ( lat::Socket *  s )

protectedpure virtual

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

Referenced by run().

debug()

void DQMNet::debug

(

bool

doit

)

Enable or disable verbose debugging. Must be called before calling run() or start().

Definition at line 945 of file DQMNet.cc.

References debug_.

Referenced by DQMService::DQMService().

{ debug_ = doit; }

delay()

void DQMNet::delay

(

int

delay

)

Set the I/O dispatching delay. Must be called before calling run() or start().

Definition at line 949 of file DQMNet.cc.

References delay_.

{ delay_ = delay; }

discard()

void DQMNet::discard ( Bucket *&  b )

staticprotected

Definition at line 62 of file DQMNet.cc.

References b, and GetRecoTauVFromDQM_MC_cff::next.

Referenced by OrderedSet.OrderedSet::pop().

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

dqmhash()

static size_t DQMNet::dqmhash ( const void *  key, size_t  keylen  )

inlinestatic

Definition at line 196 of file DQMNet.h.

References a, b, c, dqmhashfinal, dqmhashmix, dqmdumpme::k, and crabWrapper::key.

Referenced by DQMImplNet< DQMNet::Object >::findObject(), DQMService::flushStandalone(), and DQMImplNet< DQMNet::Object >::makeObject().

                                                        {
    // Reduced version of Bob Jenkins' hash function at:
    //   http://www.burtleburtle.net/bob/c/lookup3.c
#define dqmhashrot(x, k) (((x) << (k)) | ((x) >> (32 - (k))))
#define dqmhashmix(a, b, c) \
  {                         \
    a -= c;                 \
    a ^= dqmhashrot(c, 4);  \
    c += b;                 \
    b -= a;                 \
    b ^= dqmhashrot(a, 6);  \
    a += c;                 \
    c -= b;                 \
    c ^= dqmhashrot(b, 8);  \
    b += a;                 \
    a -= c;                 \
    a ^= dqmhashrot(c, 16); \
    c += b;                 \
    b -= a;                 \
    b ^= dqmhashrot(a, 19); \
    a += c;                 \
    c -= b;                 \
    c ^= dqmhashrot(b, 4);  \
    b += a;                 \
  }
#define dqmhashfinal(a, b, c) \
  {                           \
    c ^= b;                   \
    c -= dqmhashrot(b, 14);   \
    a ^= c;                   \
    a -= dqmhashrot(c, 11);   \
    b ^= a;                   \
    b -= dqmhashrot(a, 25);   \
    c ^= b;                   \
    c -= dqmhashrot(b, 16);   \
    a ^= c;                   \
    a -= dqmhashrot(c, 4);    \
    b ^= a;                   \
    b -= dqmhashrot(a, 14);   \
    c ^= b;                   \
    c -= dqmhashrot(b, 24);   \
  }

    uint32_t a, b, c;
    a = b = c = 0xdeadbeef + (uint32_t)keylen;
    const auto *k = (const unsigned char *)key;

    // all but the last block: affect some bits of (a, b, c)
    while (keylen > 12) {
      a += k[0];
      a += ((uint32_t)k[1]) << 8;
      a += ((uint32_t)k[2]) << 16;
      a += ((uint32_t)k[3]) << 24;
      b += k[4];
      b += ((uint32_t)k[5]) << 8;
      b += ((uint32_t)k[6]) << 16;
      b += ((uint32_t)k[7]) << 24;
      c += k[8];
      c += ((uint32_t)k[9]) << 8;
      c += ((uint32_t)k[10]) << 16;
      c += ((uint32_t)k[11]) << 24;
      dqmhashmix(a, b, c);
      keylen -= 12;
      k += 12;
    }

    // last block: affect all 32 bits of (c); all case statements fall through
    switch (keylen) {
      case 12:
        c += ((uint32_t)k[11]) << 24;
        [[fallthrough]];
      case 11:
        c += ((uint32_t)k[10]) << 16;
        [[fallthrough]];
      case 10:
        c += ((uint32_t)k[9]) << 8;
        [[fallthrough]];
      case 9:
        c += k[8];
        [[fallthrough]];
      case 8:
        b += ((uint32_t)k[7]) << 24;
        [[fallthrough]];
      case 7:
        b += ((uint32_t)k[6]) << 16;
        [[fallthrough]];
      case 6:
        b += ((uint32_t)k[5]) << 8;
        [[fallthrough]];
      case 5:
        b += k[4];
        [[fallthrough]];
      case 4:
        a += ((uint32_t)k[3]) << 24;
        [[fallthrough]];
      case 3:
        a += ((uint32_t)k[2]) << 16;
        [[fallthrough]];
      case 2:
        a += ((uint32_t)k[1]) << 8;
        [[fallthrough]];
      case 1:
        a += k[0];
        break;
      case 0:
        return c;
    }

    dqmhashfinal(a, b, c);
    return c;
#undef dqmhashrot
#undef dqmhashmix
#undef dqmhashfinal
  }

findObject()

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

protectedpure virtual

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

Referenced by run().

getPeer()

virtual Peer* DQMNet::getPeer ( lat::Socket *  s )

protectedpure virtual

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

listenToCollector()

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 1033 of file DQMNet.cc.

References query::host, DQMNet::AutoPeer::host, logme(), query::port, 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;
}

lock()

void DQMNet::lock

(

)

Acquire a lock on the DQM net layer.

Definition at line 1066 of file DQMNet.cc.

References communicate_, and lock_.

Referenced by DQMService::flushStandalone(), and run().

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

logme()

std::ostream & DQMNet::logme ( )

protected

Definition at line 50 of file DQMNet.cc.

References gather_cfg::cout, submitPVValidationJobs::now, and RecoSummaryTask_cfi::Time.

Referenced by listenToCollector(), run(), DQMImplNet< DQMNet::Object >::sendObjectListToPeers(), start(), startLocalServer(), and updateToCollector().

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

losePeer()

void DQMNet::losePeer ( const char *  reason, Peer *  peer, lat::IOSelectEvent *  event, lat::Error *  err = nullptr  )

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 74 of file DQMNet.cc.

References DQMNet::Peer::automatic, MillePedeFileConverter_cfg::e, submitPVResolutionJobs::err, makeMEIFBenchmarkPlots::ev, mps_fire::i, fileCollector::logme(), DQMNet::AutoPeer::peer, DQMNet::Peer::peeraddr, PixelMapPlotter::reason, alignCSCRings::s, DQMNet::Peer::sendq, DQMNet::Peer::socket, and AlCaHLTBitMon_QueryRunRegistry::string.

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

  Socket *s = peer->socket;

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

  if (ev)
    ev->source = nullptr;

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

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

makeObject()

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

protectedpure virtual

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

markObjectsDead()

virtual void DQMNet::markObjectsDead ( Peer *  p )

protectedpure virtual

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

onLocalNotify()

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 864 of file DQMNet.cc.

References visDQMUpload::buf, MillePedeFileConverter_cfg::e, makeMEIFBenchmarkPlots::ev, fileCollector::logme(), and GetRecoTauVFromDQM_MC_cff::next.

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

onMessage()

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

protectedvirtual

Definition at line 441 of file DQMNet.cc.

References cms::cuda::assert(), data, HLT_2022v12_cff::flags, if(), fileCollector::logme(), mps_check::msg, Skims_PA_cff::name, EcalTangentSkim_cfg::o, AlCaHLTBitMon_ParallelJobs::p, and AlCaHLTBitMon_QueryRunRegistry::string.

                                                                            {
  // 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 != 2 * sizeof(uint32_t)) {
        logme() << "ERROR: corrupt 'UPDATE_ME' message of length " << len << " from peer " << p->peeraddr << std::endl;
        return false;
      }

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

      p->update = true;
    }
      return true;

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

      // Send over current status: list of known objects.
      sendObjectListToPeer(msg, true, false);
    }
      return true;

    case DQM_MSG_GET_OBJECT: {
      if (debug_)
        logme() << "DEBUG: received message 'GET OBJECT' from peer " << p->peeraddr << ", size " << len << 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;
      }

      std::string name((char *)data + 3 * sizeof(uint32_t), namelen);
      Peer *owner = nullptr;
      Object *o = findObject(nullptr, name, &owner);
      if (o) {
        o->lastreq = Time::current().ns();
        if ((o->rawdata.empty() || (o->flags & DQM_PROP_STALE)) &&
            (o->flags & DQM_PROP_TYPE_MASK) > DQM_PROP_TYPE_SCALAR)
          waitForData(p, name, "", owner);
        else
          sendObjectToPeer(msg, *o, true);
      } 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());
      }
    }
      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;
      }

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

      if (debug_)
        logme() << "DEBUG: received message 'LIST BEGIN " << (flags ? "FULL" : "INCREMENTAL") << "' from "
                << p->peeraddr << ", size " << len << std::endl;

      // If we are about to receive a full list of objects, flag all
      // objects as possibly dead.  Subsequent object notifications
      // will undo this for the live objects.  We cannot delete
      // objects quite yet, as we may get inquiry from another client
      // while we are processing the incoming list, so we keep the
      // objects tentatively alive as long as we've not seen the end.
      if (flags)
        markObjectsDead(p);
    }
      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, now purge all dead
      // objects. 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)
        purgeDeadObjects(p);

      if (debug_)
        logme() << "DEBUG: received message 'LIST END " << (flags ? "FULL" : "INCREMENTAL") << "' from " << p->peeraddr
                << ", size " << len << 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[9];
      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[6];
      uint32_t &datalen = words[7];
      uint32_t &qlen = words[8];

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

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

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

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

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

      o->flags = words[2] | DQM_PROP_NEW | DQM_PROP_RECEIVED;
      o->tag = words[5];
      o->version = ((uint64_t)words[4] << 32 | words[3]);
      o->scalar.clear();
      o->qdata.clear();
      if ((o->flags & DQM_PROP_TYPE_MASK) <= DQM_PROP_TYPE_SCALAR) {
        o->rawdata.clear();
        o->scalar.insert(o->scalar.end(), objdata, qdata);
      } else if (datalen) {
        o->rawdata.clear();
        o->rawdata.insert(o->rawdata.end(), objdata, qdata);
      } else if (!o->rawdata.empty())
        o->flags |= DQM_PROP_STALE;
      o->qdata.insert(o->qdata.end(), qdata, enddata);

      // If we had an object for this one already and this is a list
      // update without data, issue an immediate data get request.
      if (o->lastreq && !datalen && (o->flags & DQM_PROP_TYPE_MASK) > DQM_PROP_TYPE_SCALAR)
        requestObjectData(p, (namelen ? &name[0] : nullptr), namelen);

      // If we have the object data, release from wait.
      if (datalen)
        releaseWaiters(name, o);
    }
      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);
      std::string name((char *)namedata, namelen);

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

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

      // If this was a known object, kill it.
      if (Object *o = findObject(p, name)) {
        o->flags |= DQM_PROP_DEAD;
        purgeDeadObjects(p);
      }

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

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

onPeerConnect()

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 819 of file DQMNet.cc.

References cms::cuda::assert(), makeMEIFBenchmarkPlots::ev, edm::storage::IORead, edm::storage::IOUrgent, DTRecHitClients_cfi::local, CommonMethods::lock(), fileCollector::logme(), onPeerData(), AlCaHLTBitMon_ParallelJobs::p, alignCSCRings::s, AlCaHLTBitMon_QueryRunRegistry::string, and relativeConstraints::value.

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.
  lock();
  Peer *p = createPeer(s);
  std::string localaddr;
  if (auto *inet = dynamic_cast<InetSocket *>(s)) {
    InetAddress peeraddr = inet->peername();
    InetAddress myaddr = inet->sockname();
    p->peeraddr = StringFormat("%1:%2").arg(peeraddr.hostname()).arg(peeraddr.port()).value();
    localaddr = StringFormat("%1:%2").arg(myaddr.hostname()).arg(myaddr.port()).value();
  } else if (auto *local = dynamic_cast<LocalSocket *>(s)) {
    p->peeraddr = local->peername().path();
    localaddr = local->sockname().path();
  } else
    assert(false);

  p->mask = IORead | IOUrgent;
  p->socket = s;

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

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

  // We are never done.
  return false;
}

onPeerData()

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

private

Handle communication to a particular client.

Definition at line 671 of file DQMNet.cc.

References cms::cuda::assert(), b, visDQMUpload::buf, data, DQMNet::Bucket::data, fileCollector::done, MillePedeFileConverter_cfg::e, makeMEIFBenchmarkPlots::ev, edm::storage::IORead, edm::storage::IOUrgent, edm::storage::IOWrite, CommonMethods::lock(), fileCollector::logme(), MESSAGE_SIZE_LIMIT, mps_check::msg, GetRecoTauVFromDQM_MC_cff::next, DQMNet::Bucket::next, AlCaHLTBitMon_ParallelJobs::p, SOCKET_READ_GROWTH, SOCKET_READ_SIZE, and validateGeometry_cfg::valid.

Referenced by onPeerConnect(), and run().

                                                  {
  lock();
  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";
      losePeer(nullptr, p, ev);
    } else
      losePeer("WARNING: lost peer connection ", p, ev);

    unlock();
    return true;
  }

  // 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) {
        losePeer("WARNING: unable to write to peer ", p, ev, &e);
        unlock();
        return true;
      }

      p->sendpos += done;
      if (p->sendpos == b->data.size()) {
        Bucket *old = p->sendq;
        p->sendq = old->next;
        p->sendpos = 0;
        old->next = nullptr;
        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) {
      auto *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.
        losePeer("WARNING: failed to read from peer ", p, ev, &e);
        unlock();
        return true;
      }
    }

    // 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) {
        losePeer("WARNING: excessively large message from ", p, ev);
        unlock();
        return true;
      }

      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 = nullptr;
          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 = nullptr;
            (*prev)->data.swap(msg.data);
          }
        }

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

        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.
  unlock();
  return false;
}

operator=()

DQMNet& DQMNet::operator= ( const DQMNet &  )

delete

packQualityData()

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

static

Pack quality results in qr into a string into for peristent storage, such as network transfer or archival.

Definition at line 158 of file DQMNet.cc.

References visDQMUpload::buf.

Referenced by DQMService::flushStandalone(), and dqm::impl::MonitorElement::packQualityData().

                                                                {
  char buf[64];
  std::ostringstream qrs;
  QReports::const_iterator qi, qe;
  for (qi = qr.begin(), qe = qr.end(); qi != qe; ++qi) {
    int pos = 0;
    sprintf(buf, "%d%c%n%.*g", qi->code, 0, &pos, DBL_DIG + 2, qi->qtresult);
    qrs << buf << '\0' << buf + pos << '\0' << qi->qtname << '\0' << qi->algorithm << '\0' << qi->message << '\0'
        << '\0';
  }
  into = qrs.str();
}

purgeDeadObjects()

virtual void DQMNet::purgeDeadObjects ( Peer *  p )

protectedpure virtual

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

releaseFromWait() [1/2]

void DQMNet::releaseFromWait ( Bucket *  msg, WaitObject &  w, Object *  o  )

protectedvirtual

Definition at line 380 of file DQMNet.cc.

References mps_check::msg, EcalTangentSkim_cfg::o, and w().

Referenced by run().

                                                                  {
  if (o)
    sendObjectToPeer(msg, *o, true);
  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());
  }
}

releaseFromWait() [2/2]

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

private

Definition at line 132 of file DQMNet.cc.

References cms::cuda::assert(), mps_fire::i, mps_check::msg, DQMNet::Bucket::next, and EcalTangentSkim_cfg::o.

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

  releaseFromWait(*msg, *i, o);

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

releaseWaiters()

void DQMNet::releaseWaiters ( const std::string &  name, Object *  o  )

private

Definition at line 147 of file DQMNet.cc.

References MillePedeFileConverter_cfg::e, mps_fire::i, Skims_PA_cff::name, and EcalTangentSkim_cfg::o.

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

removePeer()

virtual void DQMNet::removePeer ( Peer *  p, lat::Socket *  s  )

protectedpure virtual

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

requestObjectData()

void DQMNet::requestObjectData ( Peer *  p, const char *  name, size_t  len  )

private

Queue an object request to the data server.

Definition at line 100 of file DQMNet.cc.

References mps_check::msg, Skims_PA_cff::name, DQMNet::Bucket::next, and AlCaHLTBitMon_ParallelJobs::p.

                                                                    {
  // Issue request to peer.
  Bucket **msg = &p->sendq;
  while (*msg)
    msg = &(*msg)->next;
  *msg = new Bucket;
  (*msg)->next = nullptr;

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

run()

void DQMNet::run

(

)

Run the actual I/O processing loop.

Definition at line 1091 of file DQMNet.cc.

References generateTowerEtThresholdLUT::addr, copydata(), createPeer(), debug_, delay_, downstream_, DQM_MSG_LIST_OBJECTS, DQM_MSG_UPDATE_ME, DQM_PROP_STALE, MillePedeFileConverter_cfg::e, findObject(), flush_, mps_fire::i, edm::storage::IORead, edm::storage::IOUrgent, edm::storage::IOWrite, lock(), logme(), DQMNet::AutoPeer::next, submitPVValidationJobs::now, EcalTangentSkim_cfg::o, onPeerData(), AlCaHLTBitMon_ParallelJobs::p, releaseFromWait(), alignCSCRings::s, sel_, sendObjectListToPeers(), shouldStop(), SOCKET_BUF_SIZE, RecoSummaryTask_cfi::Time, unlock(), updatePeerMasks(), upstream_, relativeConstraints::value, waiting_, waitMax_, and waitStale_.

Referenced by Types.EventID::cppID(), and Types.LuminosityBlockID::cppID().

                 {
  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 (auto ap : automatic) {
      // 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 = nullptr;
        try {
          InetAddress addr(ap->host.c_str(), ap->port);
          s = new InetSocket(SOCK_STREAM, 0, addr.family());
          s->setBlocking(false);
          s->connect(addr);
          s->setopt(SO_SNDBUF, SOCKET_BUF_SIZE);
          s->setopt(SO_RCVBUF, SOCKET_BUF_SIZE);
        } catch (Error &e) {
          auto *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 (s)
              s->abort();
            delete s;
            s = nullptr;
          }
        }

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

          InetAddress peeraddr = ((InetSocket *)s)->peername();
          InetAddress myaddr = ((InetSocket *)s)->sockname();
          p->peeraddr = StringFormat("%1:%2").arg(peeraddr.hostname()).arg(peeraddr.port()).value();
          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[4] = {2 * sizeof(uint32_t), DQM_MSG_LIST_OBJECTS, 2 * sizeof(uint32_t), DQM_MSG_UPDATE_ME};
            p->sendq = new Bucket;
            p->sendq->next = nullptr;
            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();
    lock();

    // Check if flush is required.  Flush only if one is needed.
    // Always sends the full object list, but only rarely.
    if (flush_ && now > nextFlush) {
      flush_ = false;
      nextFlush = now + TimeSpan(0, 0, 0, 15 /* seconds */, 0);
      sendObjectListToPeers(true);
    }

    // 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.
    Time waitold = now - waitMax_;
    Time waitstale = now - waitStale_;
    for (auto i = waiting_.begin(), e = waiting_.end(); i != e;) {
      Object *o = findObject(nullptr, i->name);

      // If we have (stale) object data, wait only up to stale limit.
      // Otherwise if we have no data at all, wait up to the max limit.
      if (i->time < waitold) {
        logme() << "WARNING: source not responding in " << (waitMax_.ns() * 1e-9) << "s to retrieval, releasing '"
                << i->name << "' from wait, have " << (o ? o->rawdata.size() : 0) << " data available\n";
        releaseFromWait(i++, o);
      } else if (i->time < waitstale && o && (o->flags & DQM_PROP_STALE)) {
        logme() << "WARNING: source not responding in " << (waitStale_.ns() * 1e-9) << "s to update, releasing '"
                << i->name << "' from wait, have " << o->rawdata.size() << " data available\n";
        releaseFromWait(i++, o);
      }

      // Keep it for now.
      else
        ++i;
    }

    unlock();
  }
}

sendLocalChanges()

void DQMNet::sendLocalChanges

(

)

Definition at line 1203 of file DQMNet.cc.

References wakeup_.

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

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

sendObjectListToPeer()

virtual void DQMNet::sendObjectListToPeer ( Bucket *  msg, bool  all, bool  clear  )

protectedpure virtual

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

sendObjectListToPeers()

virtual void DQMNet::sendObjectListToPeers ( bool  all )

protectedpure virtual

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

Referenced by run().

sendObjectToPeer()

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

protectedvirtual

Definition at line 398 of file DQMNet.cc.

References data, HLT_2022v12_cff::flags, mps_check::msg, and EcalTangentSkim_cfg::o.

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

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

  if ((flags & DQM_PROP_TYPE_MASK) <= DQM_PROP_TYPE_SCALAR)
    objdata.insert(objdata.end(), &o.scalar[0], &o.scalar[0] + o.scalar.size());
  else if (data)
    objdata.insert(objdata.end(), &o.rawdata[0], &o.rawdata[0] + o.rawdata.size());

  uint32_t words[9];
  uint32_t namelen = o.dirname.size() + o.objname.size() + 1;
  uint32_t datalen = objdata.size();
  uint32_t qlen = o.qdata.size();

  if (o.dirname.empty())
    --namelen;

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

  msg->data.reserve(msg->data.size() + words[0]);
  copydata(msg, &words[0], 9 * sizeof(uint32_t));
  if (namelen) {
    copydata(msg, &(o.dirname)[0], o.dirname.size());
    if (!o.dirname.empty())
      copydata(msg, "/", 1);
    copydata(msg, &o.objname[0], o.objname.size());
  }
  if (datalen)
    copydata(msg, &objdata[0], datalen);
  if (qlen)
    copydata(msg, &o.qdata[0], qlen);
}

setOrder()

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

inlinestatic

Definition at line 167 of file DQMNet.h.

References a, and b.

Referenced by dqm::impl::MonitorElement::operator<().

                                                                 {
    if (a.run == b.run) {
      if (a.lumi == b.lumi) {
        if (a.streamId == b.streamId) {
          if (a.moduleId == b.moduleId) {
            if (a.dirname == b.dirname) {
              return a.objname < b.objname;
            }
            return a.dirname < b.dirname;
          }
          return a.moduleId < b.moduleId;
        }
        return a.streamId < b.streamId;
      }
      return a.lumi < b.lumi;
    }
    return a.run < b.run;
  }

shouldStop()

bool DQMNet::shouldStop ( )

protectedvirtual

Definition at line 376 of file DQMNet.cc.

Referenced by run().

{ return shutdown_; }

shutdown()

void DQMNet::shutdown

(

)

Stop the network layer and wait it to finish.

Definition at line 1046 of file DQMNet.cc.

References communicate_, and shutdown_.

Referenced by DQMService::shutdown().

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

staleObjectWaitLimit()

void DQMNet::staleObjectWaitLimit

(

lat::TimeSpan

time

)

Set the time limit for waiting updates to stale objects. Once limit has been exhausted whatever data exists is returned. Applies only when data has been received, another time limit is applied when no data payload has been received at all.

Definition at line 955 of file DQMNet.cc.

References protons_cff::time, and waitStale_.

{ waitStale_ = time; }

start()

void DQMNet::start

(

)

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 1080 of file DQMNet.cc.

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

Referenced by progressbar.ProgressBar::__next__(), Types.LuminosityBlockRange::cppID(), Types.EventRange::cppID(), and DQMService::DQMService().

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

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

startLocalServer() [1/2]

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 960 of file DQMNet.cc.

References generateTowerEtThresholdLUT::addr, MillePedeFileConverter_cfg::e, Exception, edm::storage::IOAccept, logme(), onPeerConnect(), query::port, alignCSCRings::s, sel_, server_, and SOCKET_BUF_SIZE.

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

  try {
    InetAddress addr("0.0.0.0", port);
    auto *s = new InetSocket(SOCK_STREAM, 0, addr.family());
    s->bind(addr);
    s->listen(10);
    s->setopt(SO_SNDBUF, SOCKET_BUF_SIZE);
    s->setopt(SO_RCVBUF, SOCKET_BUF_SIZE);
    s->setBlocking(false);
    sel_.attach(server_ = s, 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;

    throw cms::Exception("DQMNet::startLocalServer") << "Failed to start server at port " <<

        port << ": " << e.explain().c_str();
  }

  logme() << "INFO: DQM server started at port " << port << std::endl;
}

startLocalServer() [2/2]

void DQMNet::startLocalServer

(

const char *

path

)

Start a server socket for accessing this DQM node over a file system socket. Must be called before calling run() or start(). May throw an Exception if the server socket cannot be initialised.

Definition at line 991 of file DQMNet.cc.

References MillePedeFileConverter_cfg::e, Exception, edm::storage::IOAccept, logme(), onPeerConnect(), castor_dqm_sourceclient_file_cfg::path, sel_, server_, and SOCKET_BUF_SIZE.

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

  try {
    server_ = new LocalServerSocket(path, 10);
    server_->setopt(SO_SNDBUF, SOCKET_BUF_SIZE);
    server_->setopt(SO_RCVBUF, 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 path " << path << ": " << e.explain() << std::endl;

    throw cms::Exception("DQMNet::startLocalServer")
        << "Failed to start server at path " << path << ": " << e.explain().c_str();
  }

  logme() << "INFO: DQM server started at path " << path << std::endl;
}

unlock()

void DQMNet::unlock

(

)

Release the lock on the DQM net layer.

Definition at line 1072 of file DQMNet.cc.

References communicate_, and lock_.

Referenced by DQMService::flushStandalone(), and run().

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

unpackQualityData()

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

static

Unpack the quality results from string from into qr. Assumes the data was saved with packQualityData().

Definition at line 173 of file DQMNet.cc.

References MonitorElementData::QReport::QValue::algorithm, MonitorElementData::QReport::QValue::code, DQM_PROP_REPORT_ERROR, DQM_PROP_REPORT_OTHER, DQM_PROP_REPORT_WARN, dqm::qstatus::ERROR, HLT_2022v12_cff::flags, MonitorElementData::QReport::QValue::message, MonitorElementData::QReport::QValue::qtname, MonitorElementData::QReport::QValue::qtresult, dqm::qstatus::STATUS_OK, and dqm::qstatus::WARNING.

                                                                              {
  const char *qdata = from;

  // Count how many qresults there are.
  size_t nqv = 0;
  while (*qdata) {
    ++nqv;
    while (*qdata)
      ++qdata;
    ++qdata;
    while (*qdata)
      ++qdata;
    ++qdata;
    while (*qdata)
      ++qdata;
    ++qdata;
    while (*qdata)
      ++qdata;
    ++qdata;
    while (*qdata)
      ++qdata;
    ++qdata;
  }

  // Now extract the qreports.
  qdata = from;
  qr.reserve(nqv);
  while (*qdata) {
    qr.emplace_back();
    DQMNet::QValue &qv = qr.back();

    qv.code = atoi(qdata);
    while (*qdata)
      ++qdata;
    switch (qv.code) {
      case dqm::qstatus::STATUS_OK:
        break;
      case dqm::qstatus::WARNING:
        flags |= DQMNet::DQM_PROP_REPORT_WARN;
        break;
      case dqm::qstatus::ERROR:
        flags |= DQMNet::DQM_PROP_REPORT_ERROR;
        break;
      default:
        flags |= DQMNet::DQM_PROP_REPORT_OTHER;
        break;
    }

    qv.qtresult = atof(++qdata);
    while (*qdata)
      ++qdata;

    qv.qtname = ++qdata;
    while (*qdata)
      ++qdata;

    qv.algorithm = ++qdata;
    while (*qdata)
      ++qdata;

    qv.message = ++qdata;
    while (*qdata)
      ++qdata;
    ++qdata;
  }
}

updateMask()

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 888 of file DQMNet.cc.

References cms::cuda::assert(), edm::storage::IOUrgent, edm::storage::IOWrite, fileCollector::logme(), and AlCaHLTBitMon_ParallelJobs::p.

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(nullptr, p, nullptr);
  }
}

updatePeerMasks()

virtual void DQMNet::updatePeerMasks ( )

protectedpure virtual

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

Referenced by run().

updateToCollector()

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 1018 of file DQMNet.cc.

References downstream_, query::host, DQMNet::AutoPeer::host, logme(), query::port, 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;
}

waitForData()

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 118 of file DQMNet.cc.

References info(), Skims_PA_cff::name, and AlCaHLTBitMon_ParallelJobs::p.

                                                                                           {
  // 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.
  requestObjectData(owner, !name.empty() ? &name[0] : nullptr, name.size());
  WaitObject wo = {Time::current(), name, info, p};
  waiting_.push_back(wo);
  p->waiting++;
}

Member Data Documentation

appname_

std::string DQMNet::appname_

private

Definition at line 354 of file DQMNet.h.

communicate_

pthread_t DQMNet::communicate_

private

Definition at line 366 of file DQMNet.h.

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

debug_

bool DQMNet::debug_

protected

Definition at line 341 of file DQMNet.h.

Referenced by debug(), run(), and DQMImplNet< DQMNet::Object >::sendObjectListToPeers().

delay_

int DQMNet::delay_

private

Definition at line 369 of file DQMNet.h.

Referenced by delay(), and run().

downstream_

AutoPeer DQMNet::downstream_

private

Definition at line 363 of file DQMNet.h.

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

DQM_MSG_GET_OBJECT

const uint32_t DQMNet::DQM_MSG_GET_OBJECT = 3

static

Definition at line 72 of file DQMNet.h.

DQM_MSG_HELLO

const uint32_t DQMNet::DQM_MSG_HELLO = 0

static

Definition at line 69 of file DQMNet.h.

DQM_MSG_LIST_OBJECTS

const uint32_t DQMNet::DQM_MSG_LIST_OBJECTS = 2

static

Definition at line 71 of file DQMNet.h.

Referenced by run().

DQM_MSG_UPDATE_ME

const uint32_t DQMNet::DQM_MSG_UPDATE_ME = 1

static

Definition at line 70 of file DQMNet.h.

Referenced by run().

DQM_PROP_ACCUMULATE

const uint32_t DQMNet::DQM_PROP_ACCUMULATE = 0x00004000

static

Definition at line 58 of file DQMNet.h.

DQM_PROP_DEAD

const uint32_t DQMNet::DQM_PROP_DEAD = 0x00080000

static

Definition at line 64 of file DQMNet.h.

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

DQM_PROP_EFFICIENCY_PLOT

const uint32_t DQMNet::DQM_PROP_EFFICIENCY_PLOT = 0x00200000

static

Definition at line 66 of file DQMNet.h.

Referenced by dqm::DQMFileSaverPB::savePB(), and dqm::impl::MonitorElement::syncCoreObject().

DQM_PROP_HAS_REFERENCE

const uint32_t DQMNet::DQM_PROP_HAS_REFERENCE = 0x00001000

static

Definition at line 56 of file DQMNet.h.

DQM_PROP_LUMI

const uint32_t DQMNet::DQM_PROP_LUMI = 0x00040000

static

Definition at line 63 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load(), dqm::DQMFileSaverPB::savePB(), and dqm::impl::MonitorElement::syncCoreObject().

DQM_PROP_MARKTODELETE

const uint32_t DQMNet::DQM_PROP_MARKTODELETE = 0x01000000

static

Definition at line 67 of file DQMNet.h.

DQM_PROP_NEW

const uint32_t DQMNet::DQM_PROP_NEW = 0x00010000

static

Definition at line 61 of file DQMNet.h.

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

DQM_PROP_RECEIVED

const uint32_t DQMNet::DQM_PROP_RECEIVED = 0x00020000

static

Definition at line 62 of file DQMNet.h.

DQM_PROP_REPORT_ALARM

const uint32_t DQMNet::DQM_PROP_REPORT_ALARM = (DQM_PROP_REPORT_ERROR | DQM_PROP_REPORT_WARN | DQM_PROP_REPORT_OTHER)

static

Definition at line 54 of file DQMNet.h.

DQM_PROP_REPORT_CLEAR

const uint32_t DQMNet::DQM_PROP_REPORT_CLEAR = 0x00000000

static

Definition at line 50 of file DQMNet.h.

DQM_PROP_REPORT_ERROR

const uint32_t DQMNet::DQM_PROP_REPORT_ERROR = 0x00000100

static

Definition at line 51 of file DQMNet.h.

Referenced by dqm::impl::MonitorElement::hasError(), dqm::impl::MonitorElement::syncCoreObject(), and unpackQualityData().

DQM_PROP_REPORT_MASK

const uint32_t DQMNet::DQM_PROP_REPORT_MASK = 0x00000f00

static

Definition at line 49 of file DQMNet.h.

DQM_PROP_REPORT_OTHER

const uint32_t DQMNet::DQM_PROP_REPORT_OTHER = 0x00000400

static

Definition at line 53 of file DQMNet.h.

Referenced by dqm::impl::MonitorElement::hasOtherReport(), dqm::impl::MonitorElement::syncCoreObject(), and unpackQualityData().

DQM_PROP_REPORT_WARN

const uint32_t DQMNet::DQM_PROP_REPORT_WARN = 0x00000200

static

Definition at line 52 of file DQMNet.h.

Referenced by dqm::impl::MonitorElement::hasWarning(), dqm::impl::MonitorElement::syncCoreObject(), and unpackQualityData().

DQM_PROP_RESET

const uint32_t DQMNet::DQM_PROP_RESET = 0x00008000

static

Definition at line 59 of file DQMNet.h.

DQM_PROP_STALE

const uint32_t DQMNet::DQM_PROP_STALE = 0x00100000

static

Definition at line 65 of file DQMNet.h.

Referenced by run().

DQM_PROP_TAGGED

const uint32_t DQMNet::DQM_PROP_TAGGED = 0x00002000

static

Definition at line 57 of file DQMNet.h.

DQM_PROP_TYPE_DATABLOB

const uint32_t DQMNet::DQM_PROP_TYPE_DATABLOB = 0x00000050

static

Definition at line 47 of file DQMNet.h.

DQM_PROP_TYPE_INT

const uint32_t DQMNet::DQM_PROP_TYPE_INT = 0x00000001

static

Definition at line 31 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_INVALID

const uint32_t DQMNet::DQM_PROP_TYPE_INVALID = 0x00000000

static

Definition at line 30 of file DQMNet.h.

DQM_PROP_TYPE_MASK

const uint32_t DQMNet::DQM_PROP_TYPE_MASK = 0x000000ff

static

Definition at line 28 of file DQMNet.h.

Referenced by dqm::impl::MonitorElement::kind(), and dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_REAL

const uint32_t DQMNet::DQM_PROP_TYPE_REAL = 0x00000002

static

Definition at line 32 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_SCALAR

const uint32_t DQMNet::DQM_PROP_TYPE_SCALAR = 0x0000000f

static

Definition at line 29 of file DQMNet.h.

DQM_PROP_TYPE_STRING

const uint32_t DQMNet::DQM_PROP_TYPE_STRING = 0x00000003

static

Definition at line 33 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH1D

const uint32_t DQMNet::DQM_PROP_TYPE_TH1D = 0x00000012

static

Definition at line 36 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH1F

const uint32_t DQMNet::DQM_PROP_TYPE_TH1F = 0x00000010

static

Definition at line 34 of file DQMNet.h.

Referenced by TnPEfficiencyClient::dqmEndJob(), and dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH1I

const uint32_t DQMNet::DQM_PROP_TYPE_TH1I = 0x00000013

static

Definition at line 37 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH1S

const uint32_t DQMNet::DQM_PROP_TYPE_TH1S = 0x00000011

static

Definition at line 35 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH2D

const uint32_t DQMNet::DQM_PROP_TYPE_TH2D = 0x00000022

static

Definition at line 40 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH2F

const uint32_t DQMNet::DQM_PROP_TYPE_TH2F = 0x00000020

static

Definition at line 38 of file DQMNet.h.

Referenced by TnPEfficiencyClient::dqmEndJob(), and dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH2I

const uint32_t DQMNet::DQM_PROP_TYPE_TH2I = 0x00000023

static

Definition at line 41 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH2S

const uint32_t DQMNet::DQM_PROP_TYPE_TH2S = 0x00000021

static

Definition at line 39 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH3D

const uint32_t DQMNet::DQM_PROP_TYPE_TH3D = 0x00000032

static

Definition at line 44 of file DQMNet.h.

DQM_PROP_TYPE_TH3F

const uint32_t DQMNet::DQM_PROP_TYPE_TH3F = 0x00000030

static

Definition at line 42 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TH3S

const uint32_t DQMNet::DQM_PROP_TYPE_TH3S = 0x00000031

static

Definition at line 43 of file DQMNet.h.

DQM_PROP_TYPE_TPROF

const uint32_t DQMNet::DQM_PROP_TYPE_TPROF = 0x00000040

static

Definition at line 45 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_PROP_TYPE_TPROF2D

const uint32_t DQMNet::DQM_PROP_TYPE_TPROF2D = 0x00000041

static

Definition at line 46 of file DQMNet.h.

Referenced by dqmservices::DQMProtobufReader::load().

DQM_REPLY_LIST_BEGIN

const uint32_t DQMNet::DQM_REPLY_LIST_BEGIN = 101

static

Definition at line 74 of file DQMNet.h.

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

DQM_REPLY_LIST_END

const uint32_t DQMNet::DQM_REPLY_LIST_END = 102

static

Definition at line 75 of file DQMNet.h.

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

DQM_REPLY_NONE

const uint32_t DQMNet::DQM_REPLY_NONE = 103

static

Definition at line 76 of file DQMNet.h.

DQM_REPLY_OBJECT

const uint32_t DQMNet::DQM_REPLY_OBJECT = 104

static

Definition at line 77 of file DQMNet.h.

flush_

bool DQMNet::flush_

private

Definition at line 372 of file DQMNet.h.

Referenced by run().

lock_

pthread_mutex_t DQMNet::lock_

protected

Definition at line 342 of file DQMNet.h.

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

MAX_PEER_WAITREQS

const uint32_t DQMNet::MAX_PEER_WAITREQS = 128

static

Definition at line 79 of file DQMNet.h.

pid_

int DQMNet::pid_

private

Definition at line 355 of file DQMNet.h.

sel_

lat::IOSelector DQMNet::sel_

private

Definition at line 357 of file DQMNet.h.

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

server_

lat::Socket* DQMNet::server_

private

Definition at line 358 of file DQMNet.h.

Referenced by startLocalServer().

shutdown_

sig_atomic_t DQMNet::shutdown_

private

Definition at line 367 of file DQMNet.h.

Referenced by shutdown().

upstream_

AutoPeer DQMNet::upstream_

private

Definition at line 362 of file DQMNet.h.

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

version_

lat::Time DQMNet::version_

private

Definition at line 360 of file DQMNet.h.

waiting_

WaitList DQMNet::waiting_

private

Definition at line 364 of file DQMNet.h.

Referenced by run().

waitMax_

lat::TimeSpan DQMNet::waitMax_

private

Definition at line 371 of file DQMNet.h.

Referenced by run().

waitStale_

lat::TimeSpan DQMNet::waitStale_

private

Definition at line 370 of file DQMNet.h.

Referenced by run(), and staleObjectWaitLimit().

wakeup_

lat::Pipe DQMNet::wakeup_

private

Definition at line 359 of file DQMNet.h.

Referenced by DQMNet(), and sendLocalChanges().