Service that publishes IS objects via the webdaq REST API. More...

#include <WebdaqInfoSvc.h>

Inheritance diagram for WebdaqInfoSvc:

Collaboration diagram for WebdaqInfoSvc:

Classes
struct	Holder
	One entry per registered IS object. More...

Public Member Functions
virtual	~WebdaqInfoSvc () noexcept override
virtual StatusCode initialize	ATLAS_NOT_THREAD_SAFE () override
virtual StatusCode	configure () override
virtual StatusCode	stop () override
virtual StatusCode	finalize () override
virtual void	handle (const Incident &incident) override
virtual bool	configure (const boost::property_tree::ptree &) override
virtual bool	prepareForRun (const boost::property_tree::ptree &) override
virtual bool	prepareWorker (const boost::property_tree::ptree &) override
virtual bool	finalizeWorker (const boost::property_tree::ptree &) override
virtual bool	finalize (const boost::property_tree::ptree &) override
virtual bool	registerObject (const std::string publishPath, std::shared_ptr< hltinterface::GenericHLTContainer > obj) override
virtual bool	releaseObject (const std::string fullName) override
	De-register an object.
virtual bool	pullInfo (const std::string &, const std::string &) override
virtual bool	beginEvent (const boost::property_tree::ptree &) override
virtual bool	endEvent (const boost::property_tree::ptree &) override
virtual std::vector< std::shared_ptr< hltinterface::GenericHLTContainer > >	queryISRegistry (const std::string &) override
virtual void	get (const std::string &, THList &) override
virtual void	getUnsummed (const std::string &, std::map< std::string, std::vector< TObject * > > &) override
virtual void	clear (const std::string &) override
virtual void	reset (const std::string &) override
virtual bool	registerTObject (const std::string &, const std::string &, TObject *) override
virtual bool	discoverTObject (const std::string &, const std::string &, TObject *&) override
virtual bool	releaseTObject (const std::string &, const std::string &) override
virtual void	clearToRelease (const std::string &) override
virtual mutex_type &	getPublicationMutex () const override
	Producers are expected to hold m_pubMutex while mutating a registered container, and across the endEvent() call itself.
virtual void	setModification (bool b) override

Private Member Functions
void	publishingTask () const
	Background loop that periodically publishes containers.
bool	publishOne (Holder &h) const
	Send the pre-built snapshot via webdaq::is::put.
std::vector< std::shared_ptr< Holder > >	holders () const
	Copy of the registered holders, taken under m_objsMtx.

Static Private Member Functions
static nlohmann::json	serializeContainer (hltinterface::GenericHLTContainer &cont)
	Serialise all supported fields into a JSON object.

Private Attributes
std::shared_ptr< hltinterface::ContainerFactory >	m_factory
	Container factory.
std::unordered_map< std::string, std::shared_ptr< Holder > >	m_objs
	Registered IS objects, keyed by full IS name (publishPath + object name).
std::mutex	m_objsMtx
std::thread	m_thread
	Background publishing thread + stop flag.
std::atomic< bool >	m_stopFlag {false}
std::string	m_partition
std::string	m_tdaqWebdaqBase
std::string	m_tdaqIsServerName
std::mutex	m_pubMutex
	Publication mutex handed out by getPublicationMutex().
Gaudi::Property< std::string >	m_isServerName
Gaudi::Property< unsigned int >	m_intervalSeconds

Detailed Description

Service that publishes IS objects via the webdaq REST API.

On initialize() the service installs itself as the hltinterface::IInfoRegister singleton and provides a default ContainerFactory. Producers (e.g. TrigLArNoiseBurstRecoAlg) keep using the hltinterface API unchanged: construct GenericHLTContainer objects via the factory, register through IInfoRegister and append values inside beginEvent / endEvent. endEvent takes a JSON snapshot of each container and a background thread ships the snapshots to the IS server via webdaq.

Only the IS half of the IInfoRegister interface is implemented, histograms are handled by WebdaqHistSvc.

Required environment variables:

TDAQ_PARTITION : the partition to publish to
TDAQ_WEBDAQ_BASE : protocol, hostname and, if required, port of the webdaq server to publish through (e.g. http://localhost:8080). Same variable used by WebdaqHistSvc.
TDAQ_IS_SERVER : (optional) name of the IS server, default is taken from the ISServerName property

Definition at line 51 of file WebdaqInfoSvc.h.

Constructor & Destructor Documentation

◆ ~WebdaqInfoSvc()

virtual WebdaqInfoSvc::~WebdaqInfoSvc ( )

inlineoverridevirtualnoexcept

Definition at line 56 of file WebdaqInfoSvc.h.

56{}

Member Function Documentation

◆ ATLAS_NOT_THREAD_SAFE()

virtual StatusCode initialize WebdaqInfoSvc::ATLAS_NOT_THREAD_SAFE ( )

overridevirtual

◆ beginEvent()

virtual bool WebdaqInfoSvc::beginEvent ( const boost::property_tree::ptree & )

inlineoverridevirtual

Definition at line 83 of file WebdaqInfoSvc.h.

83{ return true; }

◆ clear()

virtual void WebdaqInfoSvc::clear ( const std::string & )

inlineoverridevirtual

Definition at line 95 of file WebdaqInfoSvc.h.

95{ NOSUPPORT_VOID(DEBUG, "Histogram clearing"); }

NOSUPPORT_VOID

#define NOSUPPORT_VOID(lvl, what)

Definition TrigServicesUtils.h:20

DEBUG

#define DEBUG

Definition page_access.h:11

◆ clearToRelease()

virtual void WebdaqInfoSvc::clearToRelease ( const std::string & )

inlineoverridevirtual

Definition at line 101 of file WebdaqInfoSvc.h.

101{ NOSUPPORT_VOID(DEBUG, "Histogram release"); }

◆ configure() [1/2]

virtual StatusCode WebdaqInfoSvc::configure ( )

inlineoverridevirtual

Definition at line 59 of file WebdaqInfoSvc.h.

59{ return StatusCode::SUCCESS; }

◆ configure() [2/2]

virtual bool WebdaqInfoSvc::configure ( const boost::property_tree::ptree & )

inlineoverridevirtual

Definition at line 67 of file WebdaqInfoSvc.h.

67{ return true; }

◆ discoverTObject()

virtual bool WebdaqInfoSvc::discoverTObject	(	const std::string &	,
		const std::string &	,
		TObject *&	)

inlineoverridevirtual

Definition at line 99 of file WebdaqInfoSvc.h.

99{ NOSUPPORT(WARNING, "Histogram discovery"); }

NOSUPPORT

#define NOSUPPORT(lvl, what)

Helper to mark unsupported interfaces.

Definition TrigServicesUtils.h:14

◆ endEvent()

bool WebdaqInfoSvc::endEvent ( const boost::property_tree::ptree & )

overridevirtual

Definition at line 207 of file WebdaqInfoSvc.cxx.

{
  // The producer has finished mutating the in-memory container. 
  // Serialise each container into its snapshot here (the producer still holds its lock), then mark it dirty so that the
  // publishing thread can ship the snapshot without touching the container itself.
  for (const auto& h : holders()) {
    try {
      nlohmann::json j = serializeContainer(*h->cont);
      {
        std::scoped_lock hlk(h->mtx);
        h->snapshot = std::move(j);
      }
      h->dirty.store(true, std::memory_order_release);
    }
    catch (const std::exception& e) {
      ATH_MSG_WARNING("Failed to serialise IS object " << h->fullName << ": " << e.what());
    }
  }
  return true;
}

◆ finalize() [1/2]

StatusCode WebdaqInfoSvc::finalize ( )

overridevirtual

Definition at line 128 of file WebdaqInfoSvc.cxx.

{
  ATH_MSG_INFO("finalize");
 
  // Uninstall ourselves.
  if (hltinterface::IInfoRegister::instance() == this) {
    hltinterface::IInfoRegister::setInstance(nullptr, /*force=*/true);
  }
 
  std::scoped_lock lk(m_objsMtx);
  m_objs.clear();
  return StatusCode::SUCCESS;
}

◆ finalize() [2/2]

virtual bool WebdaqInfoSvc::finalize ( const boost::property_tree::ptree & )

inlineoverridevirtual

Definition at line 71 of file WebdaqInfoSvc.h.

71{ return true; }

◆ finalizeWorker()

virtual bool WebdaqInfoSvc::finalizeWorker ( const boost::property_tree::ptree & )

inlineoverridevirtual

Definition at line 70 of file WebdaqInfoSvc.h.

70{ return true; }

◆ get()

virtual void WebdaqInfoSvc::get	(	const std::string &	,
		THList &	)

inlineoverridevirtual

Definition at line 92 of file WebdaqInfoSvc.h.

92{ NOSUPPORT_VOID(DEBUG, "Histogram retrieval"); }

◆ getPublicationMutex()

virtual mutex_type & WebdaqInfoSvc::getPublicationMutex ( ) const

inlineoverridevirtual

Producers are expected to hold m_pubMutex while mutating a registered container, and across the endEvent() call itself.

endEvent() serialises every registered container, because the IInfoRegister interface gives no way to tell which one the caller just finished writing.

TrigLArNoiseBurstRecoAlg is currently not using this mutex (note that this is currently not a problem since TrigLArNoiseBurstRecoAlg is the only producer we have). The reason is that MonSvcInfoService hands out the global OH mutex, and this would coupel IS publication with the OH one. In this new service we hand out our own, avoiding coupling IS to histogram operations.

Todo: Adopt this in TrigLArNoiseBurstRecoAlg, which still uses a private mutex, once the legacy service is retired. TrigExISPublishing already uses it. The mutex would become unnecessary overloading endEvent(ptree, fullName) in hltinterface, identifying the caller's object. Each producer would then touch only its own container and a private mutex would suffices. A cleaner alternative would be locking inside GenericHLTContainer. But this would require changing the accessors that are currently handing out raw references The hltinterface change should also drop const from getPublicationMutex(): returning a non-const reference from a const method forces the workaround below.

Definition at line 115 of file WebdaqInfoSvc.h.

                                                           {
    // We want to give non-const access through a const interface (IInfoRegister declares this const) --> need annotated local
    mutex_type& mtx ATLAS_THREAD_SAFE = m_pubMutex;
    return mtx;
  }

◆ getUnsummed()

virtual void WebdaqInfoSvc::getUnsummed	(	const std::string &	,
		std::map< std::string, std::vector< TObject * > > &	)

inlineoverridevirtual

Definition at line 93 of file WebdaqInfoSvc.h.

94 { NOSUPPORT_VOID(DEBUG, "Histogram retrieval"); }

◆ handle()

void WebdaqInfoSvc::handle ( const Incident & incident )

overridevirtual

Definition at line 93 of file WebdaqInfoSvc.cxx.

{
  if (incident.type() == AthenaInterprocess::UpdateAfterFork::type()) {
    ATH_MSG_INFO("Going to initialize the IS publishing thread (interval: "
                 << m_intervalSeconds.value() << " s)");
    m_thread = std::thread(&WebdaqInfoSvc::publishingTask, this);
  }
}

◆ holders()

auto WebdaqInfoSvc::holders ( ) const

private

Copy of the registered holders, taken under m_objsMtx.

Definition at line 194 of file WebdaqInfoSvc.cxx.

{
  std::vector<std::shared_ptr<Holder>> all;
  std::scoped_lock lk(m_objsMtx);
  all.reserve(m_objs.size());
  for (const auto& [_, h] : m_objs) {
    all.push_back(h);
  }
  return all;
}

◆ prepareForRun()

virtual bool WebdaqInfoSvc::prepareForRun ( const boost::property_tree::ptree & )

inlineoverridevirtual

Definition at line 68 of file WebdaqInfoSvc.h.

68{ return true; }

◆ prepareWorker()

virtual bool WebdaqInfoSvc::prepareWorker ( const boost::property_tree::ptree & )

inlineoverridevirtual

Definition at line 69 of file WebdaqInfoSvc.h.

69{ return true; }

◆ publishingTask()

void WebdaqInfoSvc::publishingTask ( ) const

private

Background loop that periodically publishes containers.

Definition at line 230 of file WebdaqInfoSvc.cxx.

{
  ATH_MSG_INFO("Started webdaq IS publishing task for partition " << m_partition
               << ", IS server " << m_tdaqIsServerName);
 
  while (!m_stopFlag) {
    // Collect the holders that have been touched since the last cycle
    std::vector<std::shared_ptr<Holder>> dirtyObjs;
    for (const auto& h : holders()) {
      if (h->dirty.exchange(false, std::memory_order_acq_rel)) {
        dirtyObjs.push_back(h);
      }
    }
    for (const auto& h : dirtyObjs) {
      // Keep the object marked dirty if the publication failed, so that it is retried next cycle
      if (!publishOne(*h)) {
        h->dirty.store(true, std::memory_order_release);
      }
      if (m_stopFlag) break;
    }
    // Contrary to WebdaqHistSvc we don't use a sync to a global period. 
    TrigServices::conditionedSleep(std::chrono::seconds(m_intervalSeconds.value()), m_stopFlag);
  }
  ATH_MSG_INFO("Webdaq IS publishing task stopped");
}

◆ publishOne()

bool WebdaqInfoSvc::publishOne ( Holder & h ) const

private

Send the pre-built snapshot via webdaq::is::put.

Definition at line 282 of file WebdaqInfoSvc.cxx.

{
  nlohmann::json j;
  {
    std::scoped_lock lk(h.mtx);
    j = h.snapshot;
  }
 
  // Registered but never filled (no endEvent yet): do not overwrite the IS object with a null value
  if (j.is_null()) {
    ATH_MSG_DEBUG("No snapshot available for IS object " << h.fullName << ", skipping publication");
    return true;
  }
 
  ATH_MSG_DEBUG("Publishing IS object " << h.fullName
                << " (type=" << h.typeName << ") to "
                << m_tdaqIsServerName << "@" << m_partition);
 
  try {
    if (!webdaq::is::put(m_partition, m_tdaqIsServerName, h.fullName,
                         h.typeName, j)) {
      ATH_MSG_WARNING("webdaq::is::put failed for " << h.fullName);
      return false;
    }
  }
  catch (const std::exception& e) {
    ATH_MSG_WARNING("webdaq::is::put threw for " << h.fullName << ": " << e.what());
    return false;
  }
  return true;
}

◆ pullInfo()

virtual bool WebdaqInfoSvc::pullInfo	(	const std::string &	,
		const std::string &	)

inlineoverridevirtual

Definition at line 81 of file WebdaqInfoSvc.h.

81{ NOSUPPORT(DEBUG, "pullInfo"); }

◆ queryISRegistry()

virtual std::vector< std::shared_ptr< hltinterface::GenericHLTContainer > > WebdaqInfoSvc::queryISRegistry ( const std::string & )

inlineoverridevirtual

Definition at line 87 of file WebdaqInfoSvc.h.

87{ NOSUPPORT(DEBUG, "queryISRegistry"); }

◆ registerObject()

bool WebdaqInfoSvc::registerObject	(	const std::string	publishPath,
		std::shared_ptr< hltinterface::GenericHLTContainer >	obj )

overridevirtual

Definition at line 145 of file WebdaqInfoSvc.cxx.

{
  if (!obj) {
    ATH_MSG_WARNING("registerObject called with null container");
    return false;
  }
 
  // Key on the full IS name: the publish path alone is not unique, producers conventionally
  // register under the same one (e.g. "/HLTObjects/") and would otherwise collide.
  const std::string full = publishPath + obj->getObjName();
 
  std::scoped_lock lk(m_objsMtx);
  if (m_objs.find(full) != m_objs.end()) {
    ATH_MSG_WARNING("IS object " << full << " is already registered");
    return false;
  }
 
  auto h = std::make_shared<Holder>();
  h->cont     = obj;
  h->fullName = full;
  h->typeName = obj->getTypeName();
  m_objs.emplace(full, std::move(h));
 
  ATH_MSG_INFO("Registered IS object " << full
               << " (type=" << obj->getTypeName() << ")"
               << " for publication on " << m_tdaqIsServerName
               << " in partition " << m_partition);
  return true;
}

◆ registerTObject()

virtual bool WebdaqInfoSvc::registerTObject	(	const std::string &	,
		const std::string &	,
		TObject *	)

inlineoverridevirtual

Definition at line 98 of file WebdaqInfoSvc.h.

98{ NOSUPPORT(WARNING, "Histogram registration"); }

◆ releaseObject()

bool WebdaqInfoSvc::releaseObject ( const std::string fullName )

overridevirtual

De-register an object.

The argument is the full IS name, i.e. the publish path passed to registerObject() followed by the container's object name.

Definition at line 179 of file WebdaqInfoSvc.cxx.

{
  std::scoped_lock lk(m_objsMtx);
  auto it = m_objs.find(fullName);
  if (it == m_objs.end()) {
    ATH_MSG_WARNING("releaseObject: " << fullName << " is not registered");
    return false;
  }
  m_objs.erase(it);
  ATH_MSG_DEBUG("Released IS object " << fullName);
  return true;
}

◆ releaseTObject()

virtual bool WebdaqInfoSvc::releaseTObject	(	const std::string &	,
		const std::string &	)

inlineoverridevirtual

Definition at line 100 of file WebdaqInfoSvc.h.

100{ NOSUPPORT(WARNING, "Histogram release"); }

◆ reset()

virtual void WebdaqInfoSvc::reset ( const std::string & )

inlineoverridevirtual

Definition at line 96 of file WebdaqInfoSvc.h.

96{ NOSUPPORT_VOID(DEBUG, "Histogram reset"); }

◆ serializeContainer()

nlohmann::json WebdaqInfoSvc::serializeContainer ( hltinterface::GenericHLTContainer & cont )

staticprivate

Serialise all supported fields into a JSON object.

Definition at line 258 of file WebdaqInfoSvc.cxx.

{
  using GHC = hltinterface::GenericHLTContainer;
  nlohmann::json j;
 
  // Only INT/FLOAT/INTVEC/FLOATVEC are supported: GenericHLTContainer::getFieldNames() throws
  // std::range_error for STRING/STRINGVEC, so those fields cannot be enumerated at all.
  auto names = cont.getFieldNames(GHC::INT);
  for (size_t i = 0; i < names.size(); ++i) j[names[i]] = cont.getIntField(i);
 
  names = cont.getFieldNames(GHC::FLOAT);
  for (size_t i = 0; i < names.size(); ++i) j[names[i]] = cont.getFloatField(i);
 
  names = cont.getFieldNames(GHC::INTVEC);
  for (size_t i = 0; i < names.size(); ++i) j[names[i]] = cont.getIntVecField(i);
 
  names = cont.getFieldNames(GHC::FLOATVEC);
  for (size_t i = 0; i < names.size(); ++i) j[names[i]] = cont.getFloatVecField(i);
 
  return j;
}

◆ setModification()

virtual void WebdaqInfoSvc::setModification ( bool b )

inlineoverridevirtual

Definition at line 120 of file WebdaqInfoSvc.h.

120{ p_setModification(b); }

◆ stop()

StatusCode WebdaqInfoSvc::stop ( )

overridevirtual

Definition at line 104 of file WebdaqInfoSvc.cxx.

{
  ATH_MSG_DEBUG("Stopping IS publishing task");
  m_stopFlag = true;
  if (m_thread.joinable()) {
    ATH_MSG_DEBUG("Going to join the IS publishing thread");
    try {
        m_thread.join();
    }
    catch (const std::exception& e) {
      ATH_MSG_ERROR("Failed to join the IS publishing thread: " << e.what());
      return StatusCode::FAILURE;
    }
  }
 
  ATH_MSG_DEBUG("Final flush of registered IS objects");
  for (const auto& h : holders()) {
    publishOne(*h);
  }
  return StatusCode::SUCCESS;
}

Member Data Documentation

◆ m_factory

std::shared_ptr<hltinterface::ContainerFactory> WebdaqInfoSvc::m_factory

private

Container factory.

Definition at line 136 of file WebdaqInfoSvc.h.

◆ m_intervalSeconds

Gaudi::Property<unsigned int> WebdaqInfoSvc::m_intervalSeconds

private

Initial value:

{this, "IntervalSeconds", 10,

"Period between publication cycles (s)"}

Definition at line 155 of file WebdaqInfoSvc.h.

155 {this, "IntervalSeconds", 10,

156 "Period between publication cycles (s)"};

◆ m_isServerName

Gaudi::Property<std::string> WebdaqInfoSvc::m_isServerName

private

Initial value:

{this, "ISServerName", "DF",

"Name of the IS server to publish to (used if TDAQ_IS_SERVER unset)"}

Definition at line 153 of file WebdaqInfoSvc.h.

153 {this, "ISServerName", "DF",

154 "Name of the IS server to publish to (used if TDAQ_IS_SERVER unset)"};

◆ m_objs

std::unordered_map<std::string, std::shared_ptr<Holder> > WebdaqInfoSvc::m_objs

private

Registered IS objects, keyed by full IS name (publishPath + object name).

Definition at line 139 of file WebdaqInfoSvc.h.

◆ m_objsMtx

std::mutex WebdaqInfoSvc::m_objsMtx

mutableprivate

Definition at line 140 of file WebdaqInfoSvc.h.

◆ m_partition

std::string WebdaqInfoSvc::m_partition

private

Definition at line 146 of file WebdaqInfoSvc.h.

◆ m_pubMutex

std::mutex WebdaqInfoSvc::m_pubMutex

mutableprivate

Publication mutex handed out by getPublicationMutex().

Definition at line 151 of file WebdaqInfoSvc.h.

◆ m_stopFlag

std::atomic<bool> WebdaqInfoSvc::m_stopFlag {false}

private

Definition at line 144 of file WebdaqInfoSvc.h.

144{false};

◆ m_tdaqIsServerName

std::string WebdaqInfoSvc::m_tdaqIsServerName

private

Definition at line 148 of file WebdaqInfoSvc.h.

◆ m_tdaqWebdaqBase

std::string WebdaqInfoSvc::m_tdaqWebdaqBase

private

Definition at line 147 of file WebdaqInfoSvc.h.

◆ m_thread

std::thread WebdaqInfoSvc::m_thread

private

Background publishing thread + stop flag.

Definition at line 143 of file WebdaqInfoSvc.h.

The documentation for this class was generated from the following files:

Classes

Public Member Functions

Private Member Functions

Static Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ ~WebdaqInfoSvc()

Member Function Documentation

◆ ATLAS_NOT_THREAD_SAFE()

◆ beginEvent()

◆ clear()

◆ clearToRelease()

◆ configure() [1/2]

◆ configure() [2/2]

◆ discoverTObject()

◆ endEvent()

◆ finalize() [1/2]

◆ finalize() [2/2]

◆ finalizeWorker()

◆ get()

◆ getPublicationMutex()

◆ getUnsummed()

◆ handle()

◆ holders()

◆ prepareForRun()

◆ prepareWorker()

◆ publishingTask()

◆ publishOne()

◆ pullInfo()

◆ queryISRegistry()

◆ registerObject()

◆ registerTObject()

◆ releaseObject()

◆ releaseTObject()

◆ reset()

◆ serializeContainer()

◆ setModification()

◆ stop()

Member Data Documentation

◆ m_factory

◆ m_intervalSeconds

◆ m_isServerName

◆ m_objs

◆ m_objsMtx

◆ m_partition

◆ m_pubMutex

◆ m_stopFlag

◆ m_tdaqIsServerName

◆ m_tdaqWebdaqBase

◆ m_thread