OnDemandMinbiasSvc Class Reference

#include <OnDemandMinbiasSvc.h>

Inheritance diagram for OnDemandMinbiasSvc:

Collaboration diagram for OnDemandMinbiasSvc:

Public Member Functions
	OnDemandMinbiasSvc (const std::string &name, ISvcLocator *svc)
	Constructor. More...
	~OnDemandMinbiasSvc () final
	Destructor. More...
StatusCode	initialize () final
	AthService initialize. More...
StatusCode	beginHardScatter (const EventContext &ctx) override
StoreGateSvc *	getMinbias (const EventContext &ctx, std::uint64_t mb_id) override
std::size_t	getNumForBunch (const EventContext &ctx, int bunch) const override
virtual std::int64_t	get_hs_id (const EventContext &ctx) const override
StatusCode	endHardScatter (const EventContext &ctx) override

Private Types
using	SGHandle = ServiceHandle< StoreGateSvc >

Private Member Functions
std::size_t	calcMBRequired (std::int64_t hs_id, std::size_t slot, unsigned int run, unsigned int lumi, std::uint64_t event)

Private Attributes
Gaudi::Property< std::uint64_t >	m_seed
Gaudi::Property< bool >	m_onDemandMB
Gaudi::Property< int >	m_skippedHSEvents
Gaudi::Property< float >	m_nPerBunch
Gaudi::Property< bool >	m_usePoisson
Gaudi::Property< bool >	m_useBeamInt
Gaudi::Property< bool >	m_useBeamLumi
Gaudi::Property< int >	m_earliestDeltaBC
Gaudi::Property< int >	m_latestDeltaBC
ServiceHandle< ISkipEventIdxSvc >	m_skipEventIdxSvc
ServiceHandle< IEvtSelector >	m_bkgEventSelector
ServiceHandle< IBeamIntensity >	m_beamInt
ServiceHandle< IBeamLuminosity >	m_beamLumi
ServiceHandle< ActiveStoreSvc >	m_activeStoreSvc
SGHandle	m_spare_store
IEvtSelector::Context *	m_bkg_evt_sel_ctx
IProxyProviderSvc *	m_proxyProviderSvc = nullptr
std::vector< std::vector< SGHandle > >	m_stores
std::vector< std::vector< std::uint64_t > >	m_num_mb_by_bunch
std::vector< std::vector< std::uint64_t > >	m_idx_lists
std::mutex	m_reading_batch_mtx
std::atomic_int64_t	m_last_loaded_hs {-1}

Static Private Attributes
static constexpr std::size_t	s_NoSlot

Detailed Description

Definition at line 29 of file OnDemandMinbiasSvc.h.

Member Typedef Documentation

SGHandle

using OnDemandMinbiasSvc::SGHandle = ServiceHandle<StoreGateSvc>

private

Definition at line 51 of file OnDemandMinbiasSvc.h.

Constructor & Destructor Documentation

OnDemandMinbiasSvc()

OnDemandMinbiasSvc::OnDemandMinbiasSvc	(	const std::string &	name,
		ISvcLocator *	svc
	)

Constructor.

Definition at line 35 of file OnDemandMinbiasSvc.cxx.

    : base_class(name, svc), m_bkg_evt_sel_ctx(nullptr), m_last_loaded_hs() {}

~OnDemandMinbiasSvc()

OnDemandMinbiasSvc::~OnDemandMinbiasSvc ( )

final

Destructor.

Definition at line 39 of file OnDemandMinbiasSvc.cxx.

{}

Member Function Documentation

beginHardScatter()

StatusCode OnDemandMinbiasSvc::beginHardScatter ( const EventContext &  ctx )

override

Definition at line 220 of file OnDemandMinbiasSvc.cxx.

                                                                       {
  using namespace std::chrono_literals;
  std::chrono::steady_clock::time_point order_wait_start{};
 
  const std::int64_t hs_id = get_hs_id(ctx);
  const std::size_t slot = ctx.slot();
  const std::size_t num_to_load =
      calcMBRequired(hs_id, slot, ctx.eventID().run_number(),
                     ctx.eventID().lumi_block(), ctx.eventID().event_number());
  auto& stores = m_stores[slot];
  // If we don't have enough stores, make more
  if (stores.size() < num_to_load) {
    ATH_MSG_INFO("Adding " << num_to_load - stores.size() << " stores");
    stores.reserve(num_to_load);
    for (std::size_t i = stores.size(); i < num_to_load; ++i) {
      auto& sg = stores.emplace_back(
          fmt::format("StoreGateSvc/StoreGate_{}_{}_{}", name(), slot, i),
          name());
      ATH_CHECK(sg.retrieve());
      sg->setStoreID(StoreID::PILEUP_STORE);
      sg->setProxyProviderSvc(m_proxyProviderSvc);
    }
  }
  // Ensure loading is done in order
  if (m_last_loaded_hs < hs_id - 1) {
    ATH_MSG_INFO("Waiting to prevent out-of-order loading. Last loaded is "
                 << m_last_loaded_hs << " and we are " << hs_id);
    order_wait_start = std::chrono::steady_clock::now();
    while (m_last_loaded_hs < hs_id - 1) {
      std::this_thread::sleep_for(50ms);
    }
    auto wait_time = std::chrono::steady_clock::now() - order_wait_start;
    ATH_MSG_INFO(fmt::format("Waited {:%M:%S} to prevent out-of-order loading",
                             wait_time));
  }
  // Lock reading mutex
  std::unique_lock lck(m_reading_batch_mtx);
  auto start = std::chrono::steady_clock::now();
  // Remember old store to reset later
  auto* old_store = m_activeStoreSvc->activeStore();
  for (std::size_t i = 0; i < num_to_load; ++i) {
    auto& sg = stores[i];
    // Change active store
    m_activeStoreSvc->setStore(sg.get());
    SG::CurrentEventStore::Push reader_sg_ces(sg.get());
    // Read next event
    ATH_CHECK(sg->clearStore(true));
    if (!(m_bkgEventSelector->next(*m_bkg_evt_sel_ctx)).isSuccess()) {
      ATH_MSG_FATAL("Ran out of minbias events");
      return StatusCode::FAILURE;
    }
    IOpaqueAddress* addr = nullptr;
    if (!m_bkgEventSelector->createAddress(*m_bkg_evt_sel_ctx, addr)
             .isSuccess()) {
      ATH_MSG_WARNING("Failed to create address. No more events?");
      return StatusCode::FAILURE;
    }
    if (addr == nullptr) {
      ATH_MSG_WARNING("createAddress returned nullptr. No more events?");
      return StatusCode::FAILURE;
    }
    ATH_CHECK(sg->recordAddress(addr));
    ATH_CHECK(sg->loadEventProxies());
    // Read data now if desired
    for (const auto* proxy_ptr : sg->proxies()) {
      if (!m_onDemandMB) {
        // Sort of a const_cast, then ->accessData()
        sg->proxy_exact(proxy_ptr->sgkey())->accessData();
      }
    }
  }
  // Reset active store
  m_activeStoreSvc->setStore(old_store);
  ATH_MSG_INFO(fmt::format("Took {:%M:%S} to load events",
                           std::chrono::steady_clock::now() - start));
  // Update last loaded
  m_last_loaded_hs.store(hs_id);
  return StatusCode::SUCCESS;
}

calcMBRequired()

std::size_t OnDemandMinbiasSvc::calcMBRequired ( std::int64_t  hs_id, std::size_t  slot, unsigned int  run, unsigned int  lumi, std::uint64_t  event  )

private

Definition at line 155 of file OnDemandMinbiasSvc.cxx.

                                                                  {
  ATH_MSG_DEBUG("Run " << run << ", lumi " << lumi << ", event " << event
                       << "| hs_id " << hs_id);
  const int n_bunches = m_latestDeltaBC.value() - m_earliestDeltaBC.value() + 1;
  // vector on stack for use if slot == s_NoSlot
  std::vector<std::uint64_t> stack_num_mb_by_bunch{};
  std::vector<std::uint64_t>& num_mb_by_bunch =
      slot == s_NoSlot ? stack_num_mb_by_bunch : m_num_mb_by_bunch[slot];
  num_mb_by_bunch.clear();
  num_mb_by_bunch.resize(n_bunches);
  FastReseededPRNG prng{m_seed.value(), hs_id};
 
  // First apply the beam luminosity SF
  bool sf_updated_throwaway;
  const float beam_lumi_sf =
      m_useBeamLumi ? m_beamLumi->scaleFactor(run, lumi, sf_updated_throwaway)
                    : 1.F;
  const float beam_lumi = beam_lumi_sf * m_nPerBunch.value();
  std::vector<float> avg_num_mb_by_bunch(n_bunches, beam_lumi);
  // Now update using beam intensities
  if (m_useBeamInt) {
    // Supposed to be once per event, but ends up running once per minbias type
    // per event now
    m_beamInt->selectT0(run, event);
    for (int bunch = m_earliestDeltaBC.value();
         bunch <= m_latestDeltaBC.value(); ++bunch) {
      std::size_t idx = bunch - m_earliestDeltaBC.value();
      avg_num_mb_by_bunch[idx] *= m_beamInt->normFactor(bunch);
    }
  }
 
  if (m_usePoisson) {
    std::transform(avg_num_mb_by_bunch.begin(), avg_num_mb_by_bunch.end(),
                   num_mb_by_bunch.begin(), [&prng](float avg) {
                     return std::poisson_distribution<std::uint64_t>(avg)(prng);
                   });
  } else {
    std::transform(avg_num_mb_by_bunch.begin(), avg_num_mb_by_bunch.end(),
                   num_mb_by_bunch.begin(), [](float f) {
                     return static_cast<std::uint64_t>(std::round(f));
                   });
  }
 
  std::uint64_t num_mb = ranges::accumulate(num_mb_by_bunch, 0UL);
  if (slot == s_NoSlot) {
    return num_mb;
  }
  // Won't go past here if slot == s_NoSlot
 
  std::vector<std::uint64_t>& index_array = m_idx_lists[slot];
  index_array.clear();
  index_array.resize(num_mb);
  std::iota(index_array.begin(), index_array.end(), 0);
  // Don't need to shuffle, since these events aren't reused
  // std::shuffle(index_array.begin(), index_array.end(), prng);
  ATH_MSG_DEBUG("HS ID " << hs_id << " uses " << num_mb << " events\n"
                         << fmt::format("\t\tBy bunch: [{}]\n",
                                        fmt::join(num_mb_by_bunch, ", ")));
  return num_mb;
}

endHardScatter()

StatusCode OnDemandMinbiasSvc::endHardScatter ( const EventContext &  ctx )

override

Definition at line 317 of file OnDemandMinbiasSvc.cxx.

                                                                     {
  // clear all stores
  for (auto&& sg : m_stores[ctx.slot()]) {
    ATH_CHECK(sg->clearStore());
  }
  return StatusCode::SUCCESS;
}

get_hs_id()

virtual std::int64_t OnDemandMinbiasSvc::get_hs_id ( const EventContext &  ctx ) const

inlineoverridevirtual

Definition at line 44 of file OnDemandMinbiasSvc.h.

                                                               {
    return m_skippedHSEvents.value() + ctx.evt();
  }

getMinbias()

StoreGateSvc * OnDemandMinbiasSvc::getMinbias ( const EventContext &  ctx, std::uint64_t  mb_id  )

override

Definition at line 300 of file OnDemandMinbiasSvc.cxx.

                                                                {
  const std::size_t slot = ctx.slot();
  const std::size_t index = m_idx_lists.at(slot).at(mb_id);
  return m_stores.at(ctx.slot()).at(index).get();
}

getNumForBunch()

std::size_t OnDemandMinbiasSvc::getNumForBunch ( const EventContext &  ctx, int  bunch  ) const

override

Definition at line 307 of file OnDemandMinbiasSvc.cxx.

                                                                {
  if (bunch < m_earliestDeltaBC.value() || bunch > m_latestDeltaBC.value()) {
    throw std::logic_error(fmt::format(
        "Tried to request bunch {} which is outside the range [{}, {}]", bunch,
        m_earliestDeltaBC.value(), m_latestDeltaBC.value()));
  }
  return m_num_mb_by_bunch.at(ctx.slot()).at(bunch - m_earliestDeltaBC.value());
}

initialize()

StatusCode OnDemandMinbiasSvc::initialize ( )

final

AthService initialize.

Definition at line 41 of file OnDemandMinbiasSvc.cxx.

                                          {
  m_stores.clear();
  ATH_CHECK(m_bkgEventSelector.retrieve());
  ATH_CHECK(m_activeStoreSvc.retrieve());
  ATH_CHECK(m_skipEventIdxSvc.retrieve());
  if (m_useBeamInt) {
    ATH_CHECK(m_beamInt.retrieve());
  }
  if (m_useBeamLumi) {
    ATH_CHECK(m_beamLumi.retrieve());
  }
  // Setup context
  if (!m_bkgEventSelector->createContext(m_bkg_evt_sel_ctx).isSuccess()) {
    ATH_MSG_ERROR("Failed to create background event selector context");
    return StatusCode::FAILURE;
  }
  ATH_CHECK(dynamic_cast<Service*>(m_bkgEventSelector.get())->start());
 
  // Setup proxy provider
  m_proxyProviderSvc = nullptr;
  ATH_CHECK(serviceLocator()->service(
      fmt::format("ProxyProviderSvc/BkgPPSvc_{}", name()), m_proxyProviderSvc,
      true));
  // Setup Address Providers
  auto* addressProvider =
      dynamic_cast<IAddressProvider*>(m_bkgEventSelector.get());
  if (addressProvider == nullptr) {
    ATH_MSG_WARNING(
        "Could not cast background event selector to IAddressProvider");
  } else {
    m_proxyProviderSvc->addProvider(addressProvider);
  }
  // AthenaPoolAddressProviderSvc
  IService* athPoolSvc = nullptr;
  ATH_CHECK(serviceLocator()->service(
      fmt::format("AthenaPoolAddressProviderSvc/BkgAPAPSvc_{}", name()),
      athPoolSvc));
  auto* athPoolAP = dynamic_cast<IAddressProvider*>(athPoolSvc);
  if (athPoolAP == nullptr) {
    ATH_MSG_WARNING(
        "Could not cast AthenaPoolAddressProviderSvc to IAddressProvider");
  } else {
    m_proxyProviderSvc->addProvider(athPoolAP);
  }
  // AddressRemappingSvc
  IService* addRemapSvc = nullptr;
  ATH_CHECK(serviceLocator()->service(
      fmt::format("AddressRemappingSvc/BkgARSvc_{}", name()), addRemapSvc));
  auto* addRemapAP = dynamic_cast<IAddressProvider*>(addRemapSvc);
  if (addRemapAP == nullptr) {
    ATH_MSG_WARNING("Could not cast AddressRemappingSvc to IAddressProvider");
  } else {
    m_proxyProviderSvc->addProvider(addRemapAP);
  }
 
  const std::size_t n_concurrent =
      Gaudi::Concurrency::ConcurrencyFlags::numConcurrentEvents();
  m_idx_lists.clear();
  m_idx_lists.resize(n_concurrent);
 
  m_num_mb_by_bunch.clear();
  m_num_mb_by_bunch.resize(n_concurrent);
 
  m_stores.clear();
  m_stores.resize(n_concurrent);
 
  const int n_stores = 50;  // Start with 50 stores per event
  // setup n_concurrent vectors of n_stores StoreGates in m_stores
  for (std::size_t i = 0; i < n_concurrent; ++i) {
    auto& sgs = m_stores[i];
    sgs.reserve(n_stores);
    for (int j = 0; j < n_stores; ++j) {
      // creates / retrieves a different StoreGateSvc for each slot
      auto& sg = sgs.emplace_back(
          fmt::format("StoreGateSvc/StoreGate_{}_{}_{}", name(), i, j), name());
      ATH_CHECK(sg.retrieve());
      sg->setStoreID(StoreID::PILEUP_STORE);
      sg->setProxyProviderSvc(m_proxyProviderSvc);
    }
  }
 
  // setup spare store for event skipping
  ATH_CHECK(m_spare_store.retrieve());
  m_spare_store->setStoreID(StoreID::PILEUP_STORE);
  m_spare_store->setProxyProviderSvc(m_proxyProviderSvc);
  auto skipEvent_callback = [this](
                                ISkipEventIdxSvc::EvtIter begin,
                                ISkipEventIdxSvc::EvtIter end) -> StatusCode {
    using namespace std::chrono_literals;
    auto* const old_store = m_activeStoreSvc->activeStore();
    m_activeStoreSvc->setStore(m_spare_store.get());
    ATH_MSG_INFO("Skipping " << end - begin << " HS events. ");
    for (auto iter = begin; iter < end; ++iter) {
      const auto& evt = *iter;
      const std::size_t n_to_skip = calcMBRequired(
          evt.evtIdx, s_NoSlot, evt.runNum, evt.lbNum, evt.evtNum);
      ATH_MSG_DEBUG("Skipping HS_ID " << evt.evtIdx << " --> skipping "
                                      << n_to_skip << " pileup events");
      for (std::size_t i = 0; i < n_to_skip; ++i) {
        if (!m_bkgEventSelector->next(*m_bkg_evt_sel_ctx).isSuccess()) {
          ATH_MSG_ERROR("Ran out of background events");
          return StatusCode::FAILURE;
        }
      }
    }
    m_last_loaded_hs.store((end - 1)->evtIdx);
    m_activeStoreSvc->setStore(old_store);
    return StatusCode::SUCCESS;
  };
  ATH_CHECK(m_skipEventIdxSvc->registerCallback(skipEvent_callback));
  ATH_MSG_INFO("Initializing ODMBSvc");
  return StatusCode::SUCCESS;
}

Member Data Documentation

m_activeStoreSvc

ServiceHandle<ActiveStoreSvc> OnDemandMinbiasSvc::m_activeStoreSvc

private

Initial value:

{
      this, "ActiveStoreSvc", "ActiveStoreSvc", "ActiveStoreSvc"}

Definition at line 85 of file OnDemandMinbiasSvc.h.

m_beamInt

ServiceHandle<IBeamIntensity> OnDemandMinbiasSvc::m_beamInt

private

Initial value:

{this, "BeamIntSvc", "FlatBM",
                                          "Beam intensity service"}

Definition at line 81 of file OnDemandMinbiasSvc.h.

m_beamLumi

ServiceHandle<IBeamLuminosity> OnDemandMinbiasSvc::m_beamLumi

private

Initial value:

{
      this, "BeamLumiSvc", "LumiProfileSvc", "Beam luminosity service"}

Definition at line 83 of file OnDemandMinbiasSvc.h.

m_bkg_evt_sel_ctx

IEvtSelector::Context* OnDemandMinbiasSvc::m_bkg_evt_sel_ctx

private

Definition at line 91 of file OnDemandMinbiasSvc.h.

m_bkgEventSelector

ServiceHandle<IEvtSelector> OnDemandMinbiasSvc::m_bkgEventSelector

private

Initial value:

{
      this, "BkgEventSelector", {}, "Event selector for minbias events"}

Definition at line 79 of file OnDemandMinbiasSvc.h.

m_earliestDeltaBC

Gaudi::Property<int> OnDemandMinbiasSvc::m_earliestDeltaBC

private

Initial value:

{
      this, "EarliestDeltaBC", -32,
      "Earliest bunch crossing to consider (as delta)"}

Definition at line 70 of file OnDemandMinbiasSvc.h.

m_idx_lists

std::vector<std::vector<std::uint64_t> > OnDemandMinbiasSvc::m_idx_lists

private

Definition at line 96 of file OnDemandMinbiasSvc.h.

m_last_loaded_hs

std::atomic_int64_t OnDemandMinbiasSvc::m_last_loaded_hs {-1}

private

Definition at line 99 of file OnDemandMinbiasSvc.h.

m_latestDeltaBC

Gaudi::Property<int> OnDemandMinbiasSvc::m_latestDeltaBC

private

Initial value:

{
      this, "LatestDeltaBC", +6,
      "Latest bunch crossing to consider (as delta)"}

Definition at line 73 of file OnDemandMinbiasSvc.h.

m_nPerBunch

Gaudi::Property<float> OnDemandMinbiasSvc::m_nPerBunch

private

Initial value:

{
      this, "AvgMBPerBunch", 0.f,
      "Average (max) number of minbias events per bunch"}

Definition at line 60 of file OnDemandMinbiasSvc.h.

m_num_mb_by_bunch

std::vector<std::vector<std::uint64_t> > OnDemandMinbiasSvc::m_num_mb_by_bunch

private

Definition at line 95 of file OnDemandMinbiasSvc.h.

m_onDemandMB

Gaudi::Property<bool> OnDemandMinbiasSvc::m_onDemandMB

private

Initial value:

{
      this, "OnDemandMB", false,
      "Should minbias event contents be read on demand"}

Definition at line 55 of file OnDemandMinbiasSvc.h.

m_proxyProviderSvc

IProxyProviderSvc* OnDemandMinbiasSvc::m_proxyProviderSvc = nullptr

private

Definition at line 92 of file OnDemandMinbiasSvc.h.

m_reading_batch_mtx

std::mutex OnDemandMinbiasSvc::m_reading_batch_mtx

private

Definition at line 98 of file OnDemandMinbiasSvc.h.

m_seed

Gaudi::Property<std::uint64_t> OnDemandMinbiasSvc::m_seed

private

Initial value:

{this, "Seed", 0,
                                        "Additional seed for PRNGs"}

Definition at line 53 of file OnDemandMinbiasSvc.h.

m_skipEventIdxSvc

ServiceHandle<ISkipEventIdxSvc> OnDemandMinbiasSvc::m_skipEventIdxSvc

private

Initial value:

{
      this, "SkipEvtIdxSvc", "SkipEventIdxSvc",
      "Skipped event index (run / lb num) provider"}

Definition at line 76 of file OnDemandMinbiasSvc.h.

m_skippedHSEvents

Gaudi::Property<int> OnDemandMinbiasSvc::m_skippedHSEvents

private

Initial value:

{this, "SkippedHSEvents", 0,
                                         "Number of skipped HS events"}

Definition at line 58 of file OnDemandMinbiasSvc.h.

m_spare_store

SGHandle OnDemandMinbiasSvc::m_spare_store

private

Initial value:

{this, "StoreGateSvc",
                         fmt::format("StoreGateSvc/discards_{}", name()),
                         "StoreGate for discarding events"}

Definition at line 88 of file OnDemandMinbiasSvc.h.

m_stores

std::vector<std::vector<SGHandle> > OnDemandMinbiasSvc::m_stores

private

Definition at line 94 of file OnDemandMinbiasSvc.h.

m_useBeamInt

Gaudi::Property<bool> OnDemandMinbiasSvc::m_useBeamInt

private

Initial value:

{
      this, "UseBeamInt", true, "Whether to use the beam intensity service"}

Definition at line 66 of file OnDemandMinbiasSvc.h.

m_useBeamLumi

Gaudi::Property<bool> OnDemandMinbiasSvc::m_useBeamLumi

private

Initial value:

{
      this, "UseBeamLumi", true, "Whether to use the beam luminosity service"}

Definition at line 68 of file OnDemandMinbiasSvc.h.

m_usePoisson

Gaudi::Property<bool> OnDemandMinbiasSvc::m_usePoisson

private

Initial value:

{this, "UsePoisson", true,
                                     "Whether to use a Poisson distribution "
                                     "(if False, use a delta distribution)"}

Definition at line 63 of file OnDemandMinbiasSvc.h.

s_NoSlot

constexpr std::size_t OnDemandMinbiasSvc::s_NoSlot

staticconstexprprivate

Initial value:

=
      std::numeric_limits<std::size_t>::max()

Definition at line 101 of file OnDemandMinbiasSvc.h.

---

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

• OnDemandMinbiasSvc.h
• OnDemandMinbiasSvc.cxx