#include <DQMNet.h>

Inheritance diagram for DQMNet:

Classes
struct	AutoPeer

struct	Bucket

struct	CoreObject

struct	HashEqual

struct	HashOp

struct	Object

struct	Peer

struct	QValue

struct	WaitObject

Public Types
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
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_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

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

Member Typedef Documentation

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

Definition at line 80 of file DQMNet.h.

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

Definition at line 81 of file DQMNet.h.

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

Definition at line 82 of file DQMNet.h.

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

Definition at line 83 of file DQMNet.h.

Constructor & Destructor Documentation

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

Definition at line 897 of file DQMNet.cc.

References downstream_, 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 ( )

virtual

Definition at line 922 of file DQMNet.cc.

                 {
   // FIXME
 }

DQMNet::DQMNet ( const DQMNet & )

delete

Member Function Documentation

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

staticprotected

Definition at line 48 of file DQMNet.cc.

References DQMNet::Bucket::data.

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

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

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

protectedpure virtual

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

Referenced by dqmhash(), and run().

void DQMNet::debug ( bool doit )

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

Definition at line 928 of file DQMNet.cc.

References debug_.

Referenced by DQMService::DQMService().

928 { debug_ = doit; }

DQMNet::debug_

bool debug_

Definition: DQMNet.h:344

void DQMNet::delay ( int delay )

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

Definition at line 932 of file DQMNet.cc.

References delay_.

932 { delay_ = delay; }

DQMNet::delay_

int delay_

Definition: DQMNet.h:372

DQMNet::delay

void delay(int delay)

Definition: DQMNet.cc:932

void DQMNet::discard ( Bucket *& b )

staticprotected

Definition at line 53 of file DQMNet.cc.

References b, GetRecoTauVFromDQM_MC_cff::next, and DQMNet::Bucket::next.

Referenced by dqmhash(), and OrderedSet.OrderedSet::pop().

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

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

inlinestatic

Definition at line 199 of file DQMNet.h.

References a, Vispa.Plugins.EdmBrowser.EdmDataAccessor::all(), b, HltBtagPostValidation_cff::c, hitfit::clear(), copydata(), createPeer(), data, discard(), dqmhashfinal, dqmhashmix, findObject(), HLT_2018_cff::flags, getPeer(), info(), dqmdumpme::k, logme(), makeObject(), markObjectsDead(), mps_check::msg, Skims_PA_cff::name, EcalTangentSkim_cfg::o, onMessage(), AlCaHLTBitMon_ParallelJobs::p, packQualityData(), purgeDeadObjects(), releaseFromWait(), removePeer(), alignCSCRings::s, sendObjectListToPeer(), sendObjectListToPeers(), sendObjectToPeer(), shouldStop(), AlCaHLTBitMon_QueryRunRegistry::string, unpackQualityData(), updateMask(), updatePeerMasks(), w, and waitForData().

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
   }

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 dqmhash(), and run().

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

protectedpure virtual

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

Referenced by dqmhash().

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

void DQMNet::lock ( )

Acquire a lock on the DQM net layer.

Definition at line 1054 of file DQMNet.cc.

References communicate_, and lock_.

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

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

std::ostream & DQMNet::logme ( )

protected

Definition at line 41 of file DQMNet.cc.

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

Referenced by dqmhash(), 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_
                    << "]: ";
 }

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

References DQMNet::Peer::automatic, MillePedeFileConverter_cfg::e, mps_fire::i, fileCollector::logme(), DQMNet::AutoPeer::peer, DQMNet::Peer::peeraddr, 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;
 }

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

protectedpure virtual

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

Referenced by dqmhash().

virtual void DQMNet::markObjectsDead ( Peer * p )

protectedpure virtual

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

Referenced by dqmhash().

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

References MillePedeFileConverter_cfg::e, 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;
 }

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

protectedvirtual

Definition at line 424 of file DQMNet.cc.

References DQMNet::Bucket::data, DQMNet::CoreObject::flags, HLT_2018_cff::flags, DQMNet::Object::lastreq, fileCollector::logme(), Skims_PA_cff::name, DQMNet::Peer::peeraddr, DQMNet::Object::qdata, DQMNet::Object::rawdata, DQMNet::Object::scalar, DQMNet::Peer::source, AlCaHLTBitMon_QueryRunRegistry::string, DQMNet::CoreObject::tag, DQMNet::Peer::update, DQMNet::Peer::updates, and DQMNet::CoreObject::version.

Referenced by dqmhash().

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

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

References IORead, IOUrgent, DTRecHitClients_cfi::local, CommonMethods::lock(), fileCollector::logme(), DQMNet::Peer::mask, onPeerData(), AlCaHLTBitMon_ParallelJobs::p, DQMNet::Peer::peeraddr, alignCSCRings::s, DQMNet::Peer::socket, 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;
 }

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

private

Handle communication to a particular client.

Definition at line 654 of file DQMNet.cc.

References DQMNet::Peer::automatic, b, data, DQMNet::Bucket::data, fileCollector::done, MillePedeFileConverter_cfg::e, DQMNet::Peer::incoming, IORead, IOUrgent, IOWrite, CommonMethods::lock(), fileCollector::logme(), DQMNet::Peer::mask, MESSAGE_SIZE_LIMIT, mps_check::msg, GetRecoTauVFromDQM_MC_cff::next, DQMNet::Bucket::next, DQMNet::Peer::peeraddr, DQMNet::Peer::sendpos, DQMNet::Peer::sendq, DQMNet::Peer::socket, SOCKET_READ_GROWTH, SOCKET_READ_SIZE, validateGeometry_cfg::valid, and DQMNet::Peer::waiting.

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

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

delete

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

Referenced by dqmhash(), 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();
 }

virtual void DQMNet::purgeDeadObjects ( Peer * p )

protectedpure virtual

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

Referenced by dqmhash().

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

protectedvirtual

Definition at line 363 of file DQMNet.cc.

References DQMNet::Bucket::data, and DQMNet::WaitObject::name.

Referenced by dqmhash(), and 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());
   }
 }

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

private

Definition at line 123 of file DQMNet.cc.

References mps_check::msg, and DQMNet::Bucket::next.

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

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

private

Definition at line 138 of file DQMNet.cc.

References MillePedeFileConverter_cfg::e, and mps_fire::i.

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

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

protectedpure virtual

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

Referenced by dqmhash().

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

private

Queue an object request to the data server.

Definition at line 91 of file DQMNet.cc.

References mps_check::msg, DQMNet::Bucket::next, and DQMNet::Peer::sendq.

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

void DQMNet::run ( )

Run the actual I/O processing loop.

Definition at line 1079 of file DQMNet.cc.

References generateTowerEtThresholdLUT::addr, DQMNet::Peer::automatic, copydata(), createPeer(), debug_, delay_, downstream_, DQM_MSG_LIST_OBJECTS, DQM_MSG_UPDATE_ME, DQM_PROP_STALE, MillePedeFileConverter_cfg::e, findObject(), DQMNet::CoreObject::flags, flush_, mps_fire::i, IORead, IOUrgent, IOWrite, lock(), logme(), DQMNet::Peer::mask, DQMNet::Bucket::next, DQMNet::AutoPeer::next, fileCollector::now, onPeerData(), AlCaHLTBitMon_ParallelJobs::p, DQMNet::Peer::peeraddr, DQMNet::Object::rawdata, releaseFromWait(), alignCSCRings::s, sel_, sendObjectListToPeers(), DQMNet::Peer::sendq, shouldStop(), DQMNet::Peer::socket, SOCKET_BUF_SIZE, RecoSummaryTask_cfi::Time, unlock(), DQMNet::Peer::update, 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();
   }
 }

void DQMNet::sendLocalChanges ( )

Definition at line 1191 of file DQMNet.cc.

References wakeup_.

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

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

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

protectedpure virtual

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

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

virtual void DQMNet::sendObjectListToPeers ( bool all )

protectedpure virtual

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

Referenced by dqmhash(), and run().

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

protectedvirtual

Definition at line 381 of file DQMNet.cc.

References DQMNet::Bucket::data, DQMNet::CoreObject::dirname, DQMNet::CoreObject::flags, HLT_2018_cff::flags, DQMNet::CoreObject::objname, DQMNet::Object::qdata, DQMNet::Object::rawdata, DQMNet::Object::scalar, DQMNet::CoreObject::tag, and DQMNet::CoreObject::version.

Referenced by dqmhash(), and 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);
 }

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

inlinestatic

Definition at line 170 of file DQMNet.h.

References DQMNet::CoreObject::dirname, DQMNet::CoreObject::lumi, DQMNet::CoreObject::moduleId, DQMNet::CoreObject::objname, DQMNet::CoreObject::run, and DQMNet::CoreObject::streamId.

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

bool DQMNet::shouldStop ( )

protectedvirtual

Definition at line 359 of file DQMNet.cc.

Referenced by dqmhash(), and run().

359 { return shutdown_; }

DQMNet::shutdown_

sig_atomic_t shutdown_

Definition: DQMNet.h:370

void DQMNet::shutdown ( )

Stop the network layer and wait it to finish.

Definition at line 1034 of file DQMNet.cc.

References communicate_, and shutdown_.

Referenced by DQMService::shutdown().

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

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

References ntuplemaker::time, and waitStale_.

938 { waitStale_ = time; }

ntuplemaker.time

time

Definition: ntuplemaker.py:310

DQMNet::waitStale_

lat::TimeSpan waitStale_

Definition: DQMNet.h:373

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

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

References generateTowerEtThresholdLUT::addr, MillePedeFileConverter_cfg::e, IOAccept, logme(), onPeerConnect(), raiseDQMError(), 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;
 
     raiseDQMError("DQMNet::startLocalServer",
                   "Failed to start server at port"
                   " %d: %s",
                   port,
                   e.explain().c_str());
   }
 
   logme() << "INFO: DQM server started at port " << port << std::endl;
 }

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

References MillePedeFileConverter_cfg::e, IOAccept, logme(), onPeerConnect(), raiseDQMError(), 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;
 
     raiseDQMError("DQMNet::startLocalServer",
                   "Failed to start server at path"
                   " %s: %s",
                   path,
                   e.explain().c_str());
   }
 
   logme() << "INFO: DQM server started at path " << path << std::endl;
 }

void DQMNet::unlock ( )

Release the lock on the DQM net layer.

Definition at line 1060 of file DQMNet.cc.

References communicate_, and lock_.

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

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

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

References DQMNet::QValue::algorithm, HltBtagPostValidation_cff::c, DQMNet::QValue::code, dqm::impl::MonitorElement::data_, DQMNet::CoreObject::dirname, DQM_PROP_REPORT_ERROR, DQM_PROP_REPORT_OTHER, DQM_PROP_REPORT_WARN, dqm::qstatus::ERROR, extractNextObject(), dqm::impl::MonitorElement::Fill(), DQMNet::CoreObject::flags, fileCollector::logme(), DQMNet::QValue::message, Skims_PA_cff::name, EcalTangentSkim_cfg::o, getGTfromDQMFile::obj, DQMNet::CoreObject::objname, DQMNet::Object::qdata, DQMNet::CoreObject::qreports, DQMNet::QValue::qtname, DQMNet::QValue::qtresult, DQMNet::Object::rawdata, DQMNet::Object::scalar, dqm::qstatus::STATUS_OK, DQMNet::CoreObject::tag, and dqm::qstatus::WARNING.

Referenced by dqmhash().

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

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

References IOUrgent, IOWrite, fileCollector::logme(), DQMNet::Peer::mask, DQMNet::Peer::peeraddr, DQMNet::Peer::sendq, DQMNet::Peer::socket, and DQMNet::Peer::waiting.

Referenced by dqmhash(), and 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);
   }
 }

virtual void DQMNet::updatePeerMasks ( )

protectedpure virtual

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

Referenced by dqmhash(), and run().

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

Tell the network layer to connect to host and port and automatically send updates whenever local DQM data changes. Must be called before calling run() or start().

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

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

References info(), Skims_PA_cff::name, and DQMNet::Peer::waiting.

Referenced by dqmhash().

                                                                                            {
   // 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

std::string DQMNet::appname_

private

Definition at line 357 of file DQMNet.h.

pthread_t DQMNet::communicate_

private

Definition at line 369 of file DQMNet.h.

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

bool DQMNet::debug_

protected

Definition at line 344 of file DQMNet.h.

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

int DQMNet::delay_

private

Definition at line 372 of file DQMNet.h.

Referenced by delay(), and run().

AutoPeer DQMNet::downstream_

private

Definition at line 366 of file DQMNet.h.

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

const uint32_t DQMNet::DQM_MSG_GET_OBJECT = 3

static

Definition at line 67 of file DQMNet.h.

const uint32_t DQMNet::DQM_MSG_HELLO = 0

static

Definition at line 64 of file DQMNet.h.

const uint32_t DQMNet::DQM_MSG_LIST_OBJECTS = 2

static

Definition at line 66 of file DQMNet.h.

Referenced by run().

const uint32_t DQMNet::DQM_MSG_UPDATE_ME = 1

static

Definition at line 65 of file DQMNet.h.

Referenced by run().

const uint32_t DQMNet::DQM_PROP_ACCUMULATE = 0x00004000

static

Definition at line 53 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_DEAD = 0x00080000

static

Definition at line 59 of file DQMNet.h.

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

const uint32_t DQMNet::DQM_PROP_EFFICIENCY_PLOT = 0x00200000

static

Definition at line 61 of file DQMNet.h.

Referenced by dqm::impl::MonitorElement::isEfficiency(), dqm::dqmstoreimpl::DQMStore::saveMonitorElementToROOT(), and dqm::impl::MonitorElement::setEfficiencyFlag().

const uint32_t DQMNet::DQM_PROP_HAS_REFERENCE = 0x00001000

static

Definition at line 51 of file DQMNet.h.

Referenced by dqm::dqmstoreimpl::DQMStore::book_(), dqm::dqmstoreimpl::DQMStore::extract(), and dqm::impl::MonitorElement::initialise().

const uint32_t DQMNet::DQM_PROP_LUMI = 0x00040000

static

Definition at line 58 of file DQMNet.h.

Referenced by dqm::impl::MonitorElement::getLumiFlag(), dqm::dqmstoreimpl::DQMStore::readFilePB(), and dqm::impl::MonitorElement::setLumiFlag().

const uint32_t DQMNet::DQM_PROP_MARKTODELETE = 0x01000000

static

Definition at line 62 of file DQMNet.h.

Referenced by dqm::impl::MonitorElement::markedToDelete(), and dqm::impl::MonitorElement::markToDelete().

const uint32_t DQMNet::DQM_PROP_NEW = 0x00010000

static

Definition at line 56 of file DQMNet.h.

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

const uint32_t DQMNet::DQM_PROP_RECEIVED = 0x00020000

static

Definition at line 57 of file DQMNet.h.

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

static

Definition at line 49 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_REPORT_CLEAR = 0x00000000

static

Definition at line 45 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_REPORT_ERROR = 0x00000100

static

Definition at line 46 of file DQMNet.h.

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

const uint32_t DQMNet::DQM_PROP_REPORT_MASK = 0x00000f00

static

Definition at line 44 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_REPORT_OTHER = 0x00000400

static

Definition at line 48 of file DQMNet.h.

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

const uint32_t DQMNet::DQM_PROP_REPORT_WARN = 0x00000200

static

Definition at line 47 of file DQMNet.h.

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

const uint32_t DQMNet::DQM_PROP_RESET = 0x00008000

static

Definition at line 54 of file DQMNet.h.

Referenced by dqm::impl::MonitorElement::resetMe(), and dqm::impl::MonitorElement::setResetMe().

const uint32_t DQMNet::DQM_PROP_STALE = 0x00100000

static

Definition at line 60 of file DQMNet.h.

Referenced by run().

const uint32_t DQMNet::DQM_PROP_TAGGED = 0x00002000

static

Definition at line 52 of file DQMNet.h.

Referenced by dqm::dqmstoreimpl::DQMStore::saveMonitorElementToROOT().

const uint32_t DQMNet::DQM_PROP_TYPE_DATABLOB = 0x00000050

static

Definition at line 42 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_INT = 0x00000001

static

Definition at line 28 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_INVALID = 0x00000000

static

Definition at line 27 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_MASK = 0x000000ff

static

Definition at line 25 of file DQMNet.h.

Referenced by dqm::impl::MonitorElement::kind().

const uint32_t DQMNet::DQM_PROP_TYPE_REAL = 0x00000002

static

Definition at line 29 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_SCALAR = 0x0000000f

static

Definition at line 26 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_STRING = 0x00000003

static

Definition at line 30 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TH1D = 0x00000012

static

Definition at line 33 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TH1F = 0x00000010

static

Definition at line 31 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TH1S = 0x00000011

static

Definition at line 32 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TH2D = 0x00000022

static

Definition at line 36 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TH2F = 0x00000020

static

Definition at line 34 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TH2S = 0x00000021

static

Definition at line 35 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TH3D = 0x00000032

static

Definition at line 39 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TH3F = 0x00000030

static

Definition at line 37 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TH3S = 0x00000031

static

Definition at line 38 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TPROF = 0x00000040

static

Definition at line 40 of file DQMNet.h.

const uint32_t DQMNet::DQM_PROP_TYPE_TPROF2D = 0x00000041

static

Definition at line 41 of file DQMNet.h.

const uint32_t DQMNet::DQM_REPLY_LIST_BEGIN = 101

static

Definition at line 69 of file DQMNet.h.

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

const uint32_t DQMNet::DQM_REPLY_LIST_END = 102

static

Definition at line 70 of file DQMNet.h.

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

const uint32_t DQMNet::DQM_REPLY_NONE = 103

static

Definition at line 71 of file DQMNet.h.

const uint32_t DQMNet::DQM_REPLY_OBJECT = 104

static

Definition at line 72 of file DQMNet.h.

bool DQMNet::flush_

private

Definition at line 375 of file DQMNet.h.

Referenced by run().

pthread_mutex_t DQMNet::lock_

protected

Definition at line 345 of file DQMNet.h.

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

const uint32_t DQMNet::MAX_PEER_WAITREQS = 128

static

Definition at line 74 of file DQMNet.h.

int DQMNet::pid_

private

Definition at line 358 of file DQMNet.h.

lat::IOSelector DQMNet::sel_

private

Definition at line 360 of file DQMNet.h.

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

lat::Socket* DQMNet::server_

private

Definition at line 361 of file DQMNet.h.

Referenced by startLocalServer().

sig_atomic_t DQMNet::shutdown_

private

Definition at line 370 of file DQMNet.h.

Referenced by shutdown().

AutoPeer DQMNet::upstream_

private

Definition at line 365 of file DQMNet.h.

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

lat::Time DQMNet::version_

private

Definition at line 363 of file DQMNet.h.

WaitList DQMNet::waiting_

private

Definition at line 367 of file DQMNet.h.

Referenced by run().

lat::TimeSpan DQMNet::waitMax_

private

Definition at line 374 of file DQMNet.h.

Referenced by run().

lat::TimeSpan DQMNet::waitStale_

private

Definition at line 373 of file DQMNet.h.

Referenced by run(), and staleObjectWaitLimit().

lat::Pipe DQMNet::wakeup_

private

Definition at line 362 of file DQMNet.h.

Referenced by DQMNet(), and sendLocalChanges().

Classes

Public Types

Public Member Functions

Static Public Member Functions

Static Public Attributes

Protected Member Functions

Static Protected Member Functions

Protected Attributes

Private Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation