XrdAdaptor::RequestManager Class Reference

#include <XrdRequestManager.h>

Classes
class	OpenHandler

Public Types
using	IOOffset = edm::storage::IOOffset
using	IOPosBuffer = edm::storage::IOPosBuffer
using	IOSize = edm::storage::IOSize

Public Member Functions
void	addConnections (cms::Exception &) const
std::shared_ptr< XrdCl::File >	getActiveFile () const
void	getActiveSourceNames (std::vector< std::string > &sources) const
void	getDisabledSourceNames (std::vector< std::string > &sources) const
const std::string &	getFilename () const
void	getPrettyActiveSourceNames (std::vector< std::string > &sources) const
std::future< IOSize >	handle (void *into, IOSize size, IOOffset off)
std::future< IOSize >	handle (std::shared_ptr< std::vector< IOPosBuffer >> iolist)
std::future< IOSize >	handle (std::shared_ptr< XrdAdaptor::ClientRequest > c_ptr)
RequestManager &	operator= (const RequestManager &)=delete
void	requestFailure (std::shared_ptr< XrdAdaptor::ClientRequest > c_ptr, XrdCl::Status &c_status)
	RequestManager (const RequestManager &)=delete
virtual	~RequestManager ()=default

Static Public Member Functions
static std::shared_ptr< RequestManager >	getInstance (const std::string &filename, XrdCl::OpenFlags::Flags flags, XrdCl::Access::Mode perms)

Static Public Attributes
static const unsigned int	XRD_DEFAULT_TIMEOUT = 3 * 60

Private Member Functions
void	broadcastRequest (const ClientRequest &, bool active)
void	checkSources (timespec &now, IOSize requestSize, std::vector< std::shared_ptr< Source >> &activeSources, std::vector< std::shared_ptr< Source >> &inactiveSources)
void	checkSourcesImpl (timespec &now, IOSize requestSize, std::vector< std::shared_ptr< Source >> &activeSources, std::vector< std::shared_ptr< Source >> &inactiveSources)
bool	compareSources (const timespec &now, unsigned a, unsigned b, std::vector< std::shared_ptr< Source >> &activeSources, std::vector< std::shared_ptr< Source >> &inactiveSources) const
virtual void	handleOpen (XrdCl::XRootDStatus &status, std::shared_ptr< Source >)
void	initialize (std::weak_ptr< RequestManager > selfref)
std::shared_ptr< Source >	pickSingleSource ()
std::string	prepareOpaqueString () const
void	queueUpdateCurrentServer (const std::string &)
void	reportSiteChange (std::vector< std::shared_ptr< Source >> const &iOld, std::vector< std::shared_ptr< Source >> const &iNew, std::string orig_site=std::string{}) const
	RequestManager (const std::string &filename, XrdCl::OpenFlags::Flags flags, XrdCl::Access::Mode perms)
void	splitClientRequest (const std::vector< IOPosBuffer > &iolist, std::vector< IOPosBuffer > &req1, std::vector< IOPosBuffer > &req2, std::vector< std::shared_ptr< Source >> const &activeSources) const
void	updateCurrentServer ()

Private Attributes
std::vector< std::shared_ptr< Source > >	m_activeSources
oneapi::tbb::concurrent_unordered_set< std::string >	m_disabledExcludeStrings
oneapi::tbb::concurrent_unordered_set< std::shared_ptr< Source >, SourceHash >	m_disabledSources
oneapi::tbb::concurrent_unordered_set< std::string >	m_disabledSourceStrings
std::uniform_real_distribution< float >	m_distribution
std::atomic< unsigned >	m_excluded_active_count
XrdCl::OpenFlags::Flags	m_flags
std::mt19937	m_generator
std::vector< std::shared_ptr< Source > >	m_inactiveSources
timespec	m_lastSourceCheck
const std::string	m_name
timespec	m_nextActiveSourceCheck
bool	m_nextInitialSourceToggle
std::shared_ptr< OpenHandler >	m_open_handler
XrdCl::Access::Mode	m_perms
int	m_redirectLimitDelayScale
std::atomic< std::string * >	m_serverToAdvertise
std::recursive_mutex	m_source_mutex
int	m_timeout
bool	searchMode

Detailed Description

Definition at line 44 of file XrdRequestManager.h.

Member Typedef Documentation

IOOffset

using XrdAdaptor::RequestManager::IOOffset = edm::storage::IOOffset

Definition at line 48 of file XrdRequestManager.h.

IOPosBuffer

using XrdAdaptor::RequestManager::IOPosBuffer = edm::storage::IOPosBuffer

Definition at line 47 of file XrdRequestManager.h.

IOSize

using XrdAdaptor::RequestManager::IOSize = edm::storage::IOSize

Definition at line 46 of file XrdRequestManager.h.

Constructor & Destructor Documentation

RequestManager() [1/2]

XrdAdaptor::RequestManager::RequestManager ( const RequestManager &  )

delete

Referenced by getInstance().

~RequestManager()

virtual XrdAdaptor::RequestManager::~RequestManager ( )

virtualdefault

RequestManager() [2/2]

RequestManager::RequestManager ( const std::string &  filename, XrdCl::OpenFlags::Flags  flags, XrdCl::Access::Mode  perms  )

private

Definition at line 169 of file XrdRequestManager.cc.

    : m_serverToAdvertise(nullptr),
      m_timeout(XRD_DEFAULT_TIMEOUT),
      m_nextInitialSourceToggle(false),
      m_redirectLimitDelayScale(1),
      m_name(filename),
      m_flags(flags),
      m_perms(perms),
      m_distribution(0, 100),
      m_excluded_active_count(0) {}

Member Function Documentation

addConnections()

void RequestManager::addConnections

(

cms::Exception &

ex

)

const

Add the list of active connections to the exception extra info.

Definition at line 573 of file XrdRequestManager.cc.

References cms::Exception::addAdditionalInfo(), getDisabledSourceNames(), getPrettyActiveSourceNames(), source, and CalibrationSummaryClient_cfi::sources.

Referenced by getActiveFile(), initialize(), XrdAdaptor::RequestManager::OpenHandler::open(), pickSingleSource(), and requestFailure().

                                                          {
  std::vector<std::string> sources;
  getPrettyActiveSourceNames(sources);
  for (auto const &source : sources) {
    ex.addAdditionalInfo("Active source: " + source);
  }
  sources.clear();
  getDisabledSourceNames(sources);
  for (auto const &source : sources) {
    ex.addAdditionalInfo("Disabled source: " + source);
  }
}

broadcastRequest()

void XrdAdaptor::RequestManager::broadcastRequest ( const ClientRequest &  , bool  active  )

private

Given a request, broadcast it to all sources. If active is true, broadcast is made to all active sources. Otherwise, broadcast is made to the inactive sources.

checkSources()

void RequestManager::checkSources ( timespec &  now, IOSize  requestSize, std::vector< std::shared_ptr< Source >> &  activeSources, std::vector< std::shared_ptr< Source >> &  inactiveSources  )

private

Check our set of active sources. If necessary, this will kick off a search for a new source. The source check is somewhat expensive so it is only done once every second.

Definition at line 371 of file XrdRequestManager.cc.

References CalibrationSummaryClient_cfi::activeSources, checkSourcesImpl(), compareSources(), m_lastSourceCheck, m_nextActiveSourceCheck, submitPVValidationJobs::now, and timeDiffMS().

Referenced by handle().

                                                                                     {
  edm::LogVerbatim("XrdAdaptorInternal") << "Time since last check " << timeDiffMS(now, m_lastSourceCheck)
                                         << "; last check " << m_lastSourceCheck.tv_sec << "; now " << now.tv_sec
                                         << "; next check " << m_nextActiveSourceCheck.tv_sec << std::endl;
  if (timeDiffMS(now, m_lastSourceCheck) > 1000) {
    {  // Be more aggressive about getting rid of very bad sources.
      compareSources(now, 0, 1, activeSources, inactiveSources);
      compareSources(now, 1, 0, activeSources, inactiveSources);
    }
    if (timeDiffMS(now, m_nextActiveSourceCheck) > 0) {
      checkSourcesImpl(now, requestSize, activeSources, inactiveSources);
    }
  }
}

checkSourcesImpl()

void RequestManager::checkSourcesImpl ( timespec &  now, IOSize  requestSize, std::vector< std::shared_ptr< Source >> &  activeSources, std::vector< std::shared_ptr< Source >> &  inactiveSources  )

private

Definition at line 429 of file XrdRequestManager.cc.

References CalibrationSummaryClient_cfi::activeSources, compareSources(), ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, m_distribution, m_generator, m_lastSourceCheck, m_nextActiveSourceCheck, m_open_handler, eostools::move(), submitPVValidationJobs::now, reportSiteChange(), source, timeDiffMS(), XRD_ADAPTOR_LONG_OPEN_DELAY, XRD_ADAPTOR_OPEN_PROBE_PERCENT, XRD_ADAPTOR_SHORT_OPEN_DELAY, and XRD_ADAPTOR_SOURCE_QUALITY_FUDGE.

Referenced by checkSources().

                                                                                         {
  bool findNewSource = false;
  if (activeSources.size() <= 1) {
    findNewSource = true;
    edm::LogInfo("XrdAdaptorLvl3")
        << "Looking for an additional source because the number of active sources is smaller than 2";
  } else if (activeSources.size() > 1) {
    edm::LogVerbatim("XrdAdaptorInternal") << "Source 0 quality " << activeSources[0]->getQuality()
                                           << ", source 1 quality " << activeSources[1]->getQuality() << std::endl;
    findNewSource |= compareSources(now, 0, 1, activeSources, inactiveSources);
    findNewSource |= compareSources(now, 1, 0, activeSources, inactiveSources);

    // NOTE: We could probably replace the copy with a better sort function.
    // However, there are typically very few sources and the correctness is more obvious right now.
    std::vector<std::shared_ptr<Source>> eligibleInactiveSources;
    eligibleInactiveSources.reserve(inactiveSources.size());
    for (const auto &source : inactiveSources) {
      if (timeDiffMS(now, source->getLastDowngrade()) > (XRD_ADAPTOR_SHORT_OPEN_DELAY - 1) * 1000) {
        eligibleInactiveSources.push_back(source);
      }
    }
    auto bestInactiveSource =
        std::min_element(eligibleInactiveSources.begin(),
                         eligibleInactiveSources.end(),
                         [](const std::shared_ptr<Source> &s1, const std::shared_ptr<Source> &s2) {
                           return s1->getQuality() < s2->getQuality();
                         });
    auto worstActiveSource = std::max_element(activeSources.cbegin(),
                                              activeSources.cend(),
                                              [](const std::shared_ptr<Source> &s1, const std::shared_ptr<Source> &s2) {
                                                return s1->getQuality() < s2->getQuality();
                                              });
    if (bestInactiveSource != eligibleInactiveSources.end() && bestInactiveSource->get()) {
      edm::LogVerbatim("XrdAdaptorInternal") << "Best inactive source: " << (*bestInactiveSource)->PrettyID()
                                             << ", quality " << (*bestInactiveSource)->getQuality();
    }
    edm::LogVerbatim("XrdAdaptorInternal") << "Worst active source: " << (*worstActiveSource)->PrettyID()
                                           << ", quality " << (*worstActiveSource)->getQuality();
    // Only upgrade the source if we only have one source and the best inactive one isn't too horrible.
    // Regardless, we will want to re-evaluate the new source quickly (within 5s).
    if ((bestInactiveSource != eligibleInactiveSources.end()) && activeSources.size() == 1 &&
        ((*bestInactiveSource)->getQuality() < 4 * activeSources[0]->getQuality())) {
      auto oldSources = activeSources;
      activeSources.push_back(*bestInactiveSource);
      reportSiteChange(oldSources, activeSources);
      for (auto it = inactiveSources.begin(); it != inactiveSources.end(); it++)
        if (it->get() == bestInactiveSource->get()) {
          inactiveSources.erase(it);
          break;
        }
    } else
      while ((bestInactiveSource != eligibleInactiveSources.end()) &&
             (*worstActiveSource)->getQuality() >
                 (*bestInactiveSource)->getQuality() + XRD_ADAPTOR_SOURCE_QUALITY_FUDGE) {
        edm::LogVerbatim("XrdAdaptorInternal")
            << "Removing " << (*worstActiveSource)->PrettyID() << " from active sources due to quality ("
            << (*worstActiveSource)->getQuality() << ") and promoting " << (*bestInactiveSource)->PrettyID()
            << " (quality: " << (*bestInactiveSource)->getQuality() << ")" << std::endl;
        (*worstActiveSource)->setLastDowngrade(now);
        for (auto it = inactiveSources.begin(); it != inactiveSources.end(); it++)
          if (it->get() == bestInactiveSource->get()) {
            inactiveSources.erase(it);
            break;
          }
        inactiveSources.emplace_back(*worstActiveSource);
        auto oldSources = activeSources;
        activeSources.erase(worstActiveSource);
        activeSources.emplace_back(std::move(*bestInactiveSource));
        reportSiteChange(oldSources, activeSources);
        eligibleInactiveSources.clear();
        for (const auto &source : inactiveSources)
          if (timeDiffMS(now, source->getLastDowngrade()) > (XRD_ADAPTOR_LONG_OPEN_DELAY - 1) * 1000)
            eligibleInactiveSources.push_back(source);
        bestInactiveSource = std::min_element(eligibleInactiveSources.begin(),
                                              eligibleInactiveSources.end(),
                                              [](const std::shared_ptr<Source> &s1, const std::shared_ptr<Source> &s2) {
                                                return s1->getQuality() < s2->getQuality();
                                              });
        worstActiveSource = std::max_element(activeSources.begin(),
                                             activeSources.end(),
                                             [](const std::shared_ptr<Source> &s1, const std::shared_ptr<Source> &s2) {
                                               return s1->getQuality() < s2->getQuality();
                                             });
      }
    if (!findNewSource && (timeDiffMS(now, m_lastSourceCheck) > 1000 * XRD_ADAPTOR_LONG_OPEN_DELAY)) {
      float r = m_distribution(m_generator);
      if (r < XRD_ADAPTOR_OPEN_PROBE_PERCENT) {
        findNewSource = true;
      }
    }
  }
  if (findNewSource) {
    m_open_handler->open();
    m_lastSourceCheck = now;
  }

  // Only aggressively look for new sources if we don't have two.
  if (activeSources.size() == 2) {
    now.tv_sec += XRD_ADAPTOR_LONG_OPEN_DELAY - XRD_ADAPTOR_SHORT_OPEN_DELAY;
  } else {
    now.tv_sec += XRD_ADAPTOR_SHORT_OPEN_DELAY;
  }
  m_nextActiveSourceCheck = now;
}

compareSources()

bool RequestManager::compareSources ( const timespec &  now, unsigned  a, unsigned  b, std::vector< std::shared_ptr< Source >> &  activeSources, std::vector< std::shared_ptr< Source >> &  inactiveSources  ) const

private

Helper function for checkSources; compares the quality of source A versus source B; if source A is significantly worse, remove it from the list of active sources.

NOTE: assumes two sources are active and the caller must already hold m_source_mutex

Definition at line 389 of file XrdRequestManager.cc.

References a, CalibrationSummaryClient_cfi::activeSources, b, XrdAdaptor::Source::getHostname(), SiStripPI::max, submitPVValidationJobs::now, reportSiteChange(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by checkSources(), and checkSourcesImpl().

                                                                                             {
  if (activeSources.size() < std::max(a, b) + 1) {
    return false;
  }
  unsigned quality_a = activeSources[a]->getQuality();
  unsigned quality_b = activeSources[b]->getQuality();
  bool findNewSource = false;
  if ((quality_a > 5130) || ((quality_a > 260) && (quality_b * 4 < quality_a))) {
    std::string hostname_a;
    Source::getHostname(activeSources[a]->ID(), hostname_a);
    if (quality_a > 5130) {
      edm::LogWarning("XrdAdaptorLvl3") << "Deactivating " << hostname_a << " from active sources because the quality ("
                                        << quality_a << ") is above 5130 and it is not the only active server";
    }
    if ((quality_a > 260) && (quality_b * 4 < quality_a)) {
      std::string hostname_b;
      Source::getHostname(activeSources[b]->ID(), hostname_b);
      edm::LogWarning("XrdAdaptorLvl3") << "Deactivating " << hostname_a << " from active sources because its quality ("
                                        << quality_a
                                        << ") is higher than 260 and 4 times larger than the other active server "
                                        << hostname_b << " (" << quality_b << ") ";
    }
    edm::LogVerbatim("XrdAdaptorInternal") << "Removing " << hostname_a << " from active sources due to poor quality ("
                                           << quality_a << " vs " << quality_b << ")" << std::endl;
    if (activeSources[a]->getLastDowngrade().tv_sec != 0) {
      findNewSource = true;
    }
    activeSources[a]->setLastDowngrade(now);
    inactiveSources.emplace_back(activeSources[a]);
    auto oldSources = activeSources;
    activeSources.erase(activeSources.begin() + a);
    reportSiteChange(oldSources, activeSources);
  }
  return findNewSource;
}

getActiveFile()

std::shared_ptr< XrdCl::File > RequestManager::getActiveFile

(

void

)

const

Return a pointer to an active file. Useful for metadata operations.

Definition at line 537 of file XrdRequestManager.cc.

References addConnections(), cms::Exception::addContext(), TauDecayModes::dec, edm::errors::FileReadError, m_activeSources, m_flags, m_name, m_perms, and m_source_mutex.

                                                             {
  std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
  if (m_activeSources.empty()) {
    edm::Exception ex(edm::errors::FileReadError);
    ex << "XrdAdaptor::RequestManager::getActiveFile(name='" << m_name << "', flags=0x" << std::hex << m_flags
       << ", permissions=0" << std::oct << m_perms << std::dec << ") => Source used after fatal exception.";
    ex.addContext("In XrdAdaptor::RequestManager::handle()");
    addConnections(ex);
    throw ex;
  }
  return m_activeSources[0]->getFileHandle();
}

getActiveSourceNames()

void RequestManager::getActiveSourceNames

(

std::vector< std::string > &

sources

)

const

Retrieve the names of the active sources (primarily meant to enable meaningful log messages).

Definition at line 550 of file XrdRequestManager.cc.

References m_activeSources, m_source_mutex, source, and CalibrationSummaryClient_cfi::sources.

                                                                             {
  std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
  sources.reserve(m_activeSources.size());
  for (auto const &source : m_activeSources) {
    sources.push_back(source->ID());
  }
}

getDisabledSourceNames()

void RequestManager::getDisabledSourceNames

(

std::vector< std::string > &

sources

)

const

Retrieve the names of the disabled sources (primarily meant to enable meaningful log messages).

Definition at line 566 of file XrdRequestManager.cc.

References m_disabledSourceStrings, source, and CalibrationSummaryClient_cfi::sources.

Referenced by addConnections().

                                                                               {
  sources.reserve(m_disabledSourceStrings.size());
  for (auto const &source : m_disabledSourceStrings) {
    sources.push_back(source);
  }
}

getFilename()

const std::string& XrdAdaptor::RequestManager::getFilename ( ) const

inline

Return current filename

Definition at line 107 of file XrdRequestManager.h.

References m_name.

{ return m_name; }

getInstance()

static std::shared_ptr<RequestManager> XrdAdaptor::RequestManager::getInstance ( const std::string &  filename, XrdCl::OpenFlags::Flags  flags, XrdCl::Access::Mode  perms  )

inlinestatic

Some of the callback handlers need a weak_ptr reference to the RequestManager. This allows the callback handler to know when it is OK to invoke RequestManager methods.

Hence, all instances need to be created through this factory function.

Definition at line 116 of file XrdRequestManager.h.

References corrVsCorr::filename, HLT_2024v13_cff::flags, instance, and RequestManager().

                                                                              {
      std::shared_ptr<RequestManager> instance(new RequestManager(filename, flags, perms));
      instance->initialize(instance);
      return instance;
    }

getPrettyActiveSourceNames()

void RequestManager::getPrettyActiveSourceNames

(

std::vector< std::string > &

sources

)

const

Definition at line 558 of file XrdRequestManager.cc.

References m_activeSources, m_source_mutex, source, and CalibrationSummaryClient_cfi::sources.

Referenced by addConnections().

                                                                                   {
  std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
  sources.reserve(m_activeSources.size());
  for (auto const &source : m_activeSources) {
    sources.push_back(source->PrettyID());
  }
}

handle() [1/3]

std::future<IOSize> XrdAdaptor::RequestManager::handle ( void *  into, IOSize  size, IOOffset  off  )

inline

Interface for handling a client request.

Definition at line 60 of file XrdRequestManager.h.

                                                                    {
      auto c_ptr = std::make_shared<XrdAdaptor::ClientRequest>(*this, into, size, off);
      return handle(c_ptr);
    }

handle() [2/3]

std::future< IOSize > XrdAdaptor::RequestManager::handle

(

std::shared_ptr< std::vector< IOPosBuffer >>

iolist

)

Definition at line 717 of file XrdRequestManager.cc.

References a, CalibrationSummaryClient_cfi::activeSources, cms::Exception::addContext(), cms::cuda::assert(), b, TauDecayModes::dec, edm::errors::FileReadError, GET_CLOCK_MONOTONIC, eostools::move(), submitPVValidationJobs::now, AlCaHLTBitMon_ParallelJobs::p, edm::CPUTimer::start(), edm::CPUTimer::stop(), and TrackValidation_cff::task.

                                                                                               {
  //Use a copy of m_activeSources and m_inactiveSources throughout this function
  // in order to avoid holding the lock a long time and causing a deadlock.
  // When the function is over we will update the values of the containers
  std::vector<std::shared_ptr<Source>> activeSources, inactiveSources;
  {
    std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
    activeSources = m_activeSources;
    inactiveSources = m_inactiveSources;
  }
  //Make sure we update changes when we leave the function
  std::shared_ptr<void *> guard(nullptr, [this, &activeSources, &inactiveSources](void *) {
    std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
    m_activeSources = std::move(activeSources);
    m_inactiveSources = std::move(inactiveSources);
  });

  updateCurrentServer();

  timespec now;
  GET_CLOCK_MONOTONIC(now);

  edm::CPUTimer timer;
  timer.start();

  if (activeSources.size() == 1) {
    auto c_ptr = std::make_shared<XrdAdaptor::ClientRequest>(*this, iolist);
    checkSources(now, c_ptr->getSize(), activeSources, inactiveSources);
    activeSources[0]->handle(c_ptr);
    return c_ptr->get_future();
  }
  // Make sure active
  else if (activeSources.empty()) {
    edm::Exception ex(edm::errors::FileReadError);
    ex << "XrdAdaptor::RequestManager::handle readv(name='" << m_name << "', flags=0x" << std::hex << m_flags
       << ", permissions=0" << std::oct << m_perms << std::dec << ") => Source used after fatal exception.";
    ex.addContext("In XrdAdaptor::RequestManager::handle()");
    addConnections(ex);
    throw ex;
  }

  assert(iolist.get());
  auto req1 = std::make_shared<std::vector<IOPosBuffer>>();
  auto req2 = std::make_shared<std::vector<IOPosBuffer>>();
  splitClientRequest(*iolist, *req1, *req2, activeSources);

  checkSources(now, req1->size() + req2->size(), activeSources, inactiveSources);
  // CheckSources may have removed a source
  if (activeSources.size() == 1) {
    auto c_ptr = std::make_shared<XrdAdaptor::ClientRequest>(*this, iolist);
    activeSources[0]->handle(c_ptr);
    return c_ptr->get_future();
  }

  std::shared_ptr<XrdAdaptor::ClientRequest> c_ptr1, c_ptr2;
  std::future<IOSize> future1, future2;
  if (!req1->empty()) {
    c_ptr1.reset(new XrdAdaptor::ClientRequest(*this, req1));
    activeSources[0]->handle(c_ptr1);
    future1 = c_ptr1->get_future();
  }
  if (!req2->empty()) {
    c_ptr2.reset(new XrdAdaptor::ClientRequest(*this, req2));
    activeSources[1]->handle(c_ptr2);
    future2 = c_ptr2->get_future();
  }
  if (!req1->empty() && !req2->empty()) {
    std::future<IOSize> task = std::async(
        std::launch::deferred,
        [](std::future<IOSize> a, std::future<IOSize> b) {
          // Wait until *both* results are available.  This is essential
          // as the callback may try referencing the RequestManager.  If one
          // throws an exception (causing the RequestManager to be destroyed by
          // XrdFile) and the other has a failure, then the recovery code will
          // reference the destroyed RequestManager.
          //
          // Unlike other places where we use shared/weak ptrs to maintain object
          // lifetime and destruction asynchronously, we *cannot* destroy the request
          // asynchronously as it is associated with a ROOT buffer.  We must wait until we
          // are guaranteed that XrdCl will not write into the ROOT buffer before we
          // can return.
          b.wait();
          a.wait();
          return b.get() + a.get();
        },
        std::move(future1),
        std::move(future2));
    timer.stop();
    //edm::LogVerbatim("XrdAdaptorInternal") << "Total time to create requests " << static_cast<int>(1000*timer.realTime()) << std::endl;
    return task;
  } else if (!req1->empty()) {
    return future1;
  } else if (!req2->empty()) {
    return future2;
  } else {  // Degenerate case - no bytes to read.
    std::promise<IOSize> p;
    p.set_value(0);
    return p.get_future();
  }
}

handle() [3/3]

std::future< IOSize > RequestManager::handle

(

std::shared_ptr< XrdAdaptor::ClientRequest >

c_ptr

)

Handle a client request. NOTE: The shared_ptr interface is required. Depending on the state of the manager, it may decide to issue multiple requests and return the first successful. In that case, some references to the client request may still be outstanding when this function returns.

Definition at line 612 of file XrdRequestManager.cc.

References CalibrationSummaryClient_cfi::activeSources, cms::cuda::assert(), checkSources(), GET_CLOCK_MONOTONIC, m_activeSources, m_inactiveSources, m_source_mutex, eostools::move(), submitPVValidationJobs::now, pickSingleSource(), and source.

                                                                                     {
  assert(c_ptr.get());
  timespec now;
  GET_CLOCK_MONOTONIC(now);
  //NOTE: can't hold lock while calling checkSources since can lead to lock inversion
  std::vector<std::shared_ptr<Source>> activeSources, inactiveSources;
  {
    std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
    activeSources = m_activeSources;
    inactiveSources = m_inactiveSources;
  }
  {
    //make sure we update values before calling pickSingelSource
    std::shared_ptr<void *> guard(nullptr, [this, &activeSources, &inactiveSources](void *) {
      std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
      m_activeSources = std::move(activeSources);
      m_inactiveSources = std::move(inactiveSources);
    });

    checkSources(now, c_ptr->getSize(), activeSources, inactiveSources);
  }

  std::shared_ptr<Source> source = pickSingleSource();
  source->handle(c_ptr);
  return c_ptr->get_future();
}

handleOpen()

void XrdAdaptor::RequestManager::handleOpen ( XrdCl::XRootDStatus &  status, std::shared_ptr< Source >  source  )

privatevirtual

Handle the file-open response

Definition at line 673 of file XrdRequestManager.cc.

References SiStripPI::min, alignCSCRings::s, source, mps_update::status, XRD_ADAPTOR_LONG_OPEN_DELAY, and XRD_ADAPTOR_SHORT_OPEN_DELAY.

                                                                                                 {
  std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
  if (status.IsOK()) {
    edm::LogVerbatim("XrdAdaptorInternal") << "Successfully opened new source: " << source->PrettyID() << std::endl;
    m_redirectLimitDelayScale = 1;
    for (const auto &s : m_activeSources) {
      if (source->ID() == s->ID()) {
        edm::LogVerbatim("XrdAdaptorInternal")
            << "Xrootd server returned excluded source " << source->PrettyID() << "; ignoring" << std::endl;
        unsigned returned_count = ++m_excluded_active_count;
        m_nextActiveSourceCheck.tv_sec += XRD_ADAPTOR_SHORT_OPEN_DELAY;
        if (returned_count >= 3) {
          m_nextActiveSourceCheck.tv_sec += XRD_ADAPTOR_LONG_OPEN_DELAY - 2 * XRD_ADAPTOR_SHORT_OPEN_DELAY;
        }
        return;
      }
    }
    for (const auto &s : m_inactiveSources) {
      if (source->ID() == s->ID()) {
        edm::LogVerbatim("XrdAdaptorInternal")
            << "Xrootd server returned excluded inactive source " << source->PrettyID() << "; ignoring" << std::endl;
        m_nextActiveSourceCheck.tv_sec += XRD_ADAPTOR_LONG_OPEN_DELAY - XRD_ADAPTOR_SHORT_OPEN_DELAY;
        return;
      }
    }
    if (m_activeSources.size() < 2) {
      auto oldSources = m_activeSources;
      m_activeSources.push_back(source);
      reportSiteChange(oldSources, m_activeSources);
      queueUpdateCurrentServer(source->ID());
    } else {
      m_inactiveSources.push_back(source);
    }
  } else {  // File-open failure - wait at least 120s before next attempt.
    edm::LogVerbatim("XrdAdaptorInternal") << "Got failure when trying to open a new source" << std::endl;
    int delayScale = 1;
    if (status.status == XrdCl::errRedirectLimit) {
      m_redirectLimitDelayScale = std::min(2 * m_redirectLimitDelayScale, 100);
      delayScale = m_redirectLimitDelayScale;
    }
    m_nextActiveSourceCheck.tv_sec += delayScale * XRD_ADAPTOR_LONG_OPEN_DELAY - XRD_ADAPTOR_SHORT_OPEN_DELAY;
  }
}

initialize()

void RequestManager::initialize ( std::weak_ptr< RequestManager >  selfref )

private

Some of the callback handlers (particularly, file-open one) will want to call back into the RequestManager. However, the XrdFile may have already thrown away the reference. Hence, we need a weak_ptr to the original object before we can initialize. This way, callback knows to not reference the RequestManager.

Definition at line 180 of file XrdRequestManager.cc.

References cms::Exception::addAdditionalInfo(), addConnections(), cms::Exception::addContext(), cms::cuda::assert(), cms::Exception::clearAdditionalInfo(), cms::Exception::clearContext(), cms::Exception::clearMessage(), TauDecayModes::dec, XrdAdaptor::Source::determineHostExcludeString(), web.browse_db::env, geometryDiff::file, edm::errors::FileOpenError, spr::find(), GET_CLOCK_MONOTONIC, XrdAdaptor::Source::getDomain(), XrdAdaptor::Source::getHostname(), XrdAdaptor::RequestManager::OpenHandler::getInstance(), XrdAdaptor::Source::getXrootdSiteFromURL(), heavyIonCSV_trainingSettings::idx, m_activeSources, m_disabledExcludeStrings, m_disabledSourceStrings, m_flags, m_lastSourceCheck, m_name, m_nextActiveSourceCheck, m_open_handler, m_perms, m_source_mutex, m_timeout, eostools::move(), pileupDistInMC::oldList, prepareOpaqueString(), queueUpdateCurrentServer(), reportSiteChange(), SendMonitoringInfo(), source, mps_update::status, AlCaHLTBitMon_QueryRunRegistry::string, updateCurrentServer(), and XRD_ADAPTOR_SHORT_OPEN_DELAY.

                                                                {
  m_open_handler = OpenHandler::getInstance(self);

  XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
  if (env) {
    env->GetInt("StreamErrorWindow", m_timeout);
  }

  std::string orig_site;
  if (!Source::getXrootdSiteFromURL(m_name, orig_site) && (orig_site.find('.') == std::string::npos)) {
    std::string hostname;
    if (Source::getHostname(orig_site, hostname)) {
      Source::getDomain(hostname, orig_site);
    }
  }

  std::unique_ptr<XrdCl::File> file;
  edm::Exception ex(edm::errors::FileOpenError);
  bool validFile = false;
  const int retries = 5;
  std::string excludeString;
  for (int idx = 0; idx < retries; idx++) {
    file = std::make_unique<XrdCl::File>();
    auto opaque = prepareOpaqueString();
    std::string new_filename =
        m_name + (!opaque.empty() ? ((m_name.find('?') == m_name.npos) ? "?" : "&") + opaque : "");
    SyncHostResponseHandler handler;
    XrdCl::XRootDStatus openStatus = file->Open(new_filename, m_flags, m_perms, &handler);
    if (!openStatus
             .IsOK()) {  // In this case, we failed immediately - this indicates we have previously tried to talk to this
      // server and it was marked bad - xrootd couldn't even queue up the request internally!
      // In practice, we obsere this happening when the call to getXrootdSiteFromURL fails due to the
      // redirector being down or authentication failures.
      ex.clearMessage();
      ex.clearContext();
      ex.clearAdditionalInfo();
      ex << "XrdCl::File::Open(name='" << m_name << "', flags=0x" << std::hex << m_flags << ", permissions=0"
         << std::oct << m_perms << std::dec << ") => error '" << openStatus.ToStr() << "' (errno=" << openStatus.errNo
         << ", code=" << openStatus.code << ")";
      ex.addContext("Calling XrdFile::open()");
      ex.addAdditionalInfo("Remote server already encountered a fatal error; no redirections were performed.");
      throw ex;
    }
    handler.WaitForResponse();
    std::unique_ptr<XrdCl::XRootDStatus> status = handler.GetStatus();
    std::unique_ptr<XrdCl::HostList> hostList = handler.GetHosts();
    tracerouteRedirections(hostList.get());
    Source::determineHostExcludeString(*file, hostList.get(), excludeString);
    assert(status);
    if (status->IsOK()) {
      validFile = true;
      break;
    } else {
      ex.clearMessage();
      ex.clearContext();
      ex.clearAdditionalInfo();
      ex << "XrdCl::File::Open(name='" << m_name << "', flags=0x" << std::hex << m_flags << ", permissions=0"
         << std::oct << m_perms << std::dec << ") => error '" << status->ToStr() << "' (errno=" << status->errNo
         << ", code=" << status->code << ")";
      ex.addContext("Calling XrdFile::open()");
      addConnections(ex);
      std::string dataServer, lastUrl;
      file->GetProperty("DataServer", dataServer);
      file->GetProperty("LastURL", lastUrl);
      if (!dataServer.empty()) {
        ex.addAdditionalInfo("Problematic data server: " + dataServer);
      }
      if (!lastUrl.empty()) {
        ex.addAdditionalInfo("Last URL tried: " + lastUrl);
        edm::LogWarning("XrdAdaptorInternal") << "Failed to open file at URL " << lastUrl << ".";
      }
      if (std::find(m_disabledSourceStrings.begin(), m_disabledSourceStrings.end(), dataServer) !=
          m_disabledSourceStrings.end()) {
        ex << ". No additional data servers were found.";
        throw ex;
      }
      if (!dataServer.empty()) {
        m_disabledSourceStrings.insert(dataServer);
        m_disabledExcludeStrings.insert(excludeString);
      }
      // In this case, we didn't go anywhere - we stayed at the redirector and it gave us a file-not-found.
      if (lastUrl == new_filename) {
        edm::LogWarning("XrdAdaptorInternal") << lastUrl << ", " << new_filename;
        throw ex;
      }
    }
  }
  if (!validFile) {
    throw ex;
  }
  SendMonitoringInfo(*file);

  timespec ts;
  GET_CLOCK_MONOTONIC(ts);

  auto source = std::make_shared<Source>(ts, std::move(file), excludeString);
  {
    std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
    auto oldList = m_activeSources;
    m_activeSources.push_back(source);
    reportSiteChange(oldList, m_activeSources, orig_site);
  }
  queueUpdateCurrentServer(source->ID());
  updateCurrentServer();

  m_lastSourceCheck = ts;
  ts.tv_sec += XRD_ADAPTOR_SHORT_OPEN_DELAY;
  m_nextActiveSourceCheck = ts;
}

operator=()

RequestManager& XrdAdaptor::RequestManager::operator= ( const RequestManager &  )

delete

pickSingleSource()

std::shared_ptr< Source > RequestManager::pickSingleSource ( )

private

Picks a single source for the next operation.

Definition at line 586 of file XrdRequestManager.cc.

References addConnections(), cms::Exception::addContext(), TauDecayModes::dec, edm::errors::FileReadError, m_activeSources, m_flags, m_name, m_nextInitialSourceToggle, m_perms, m_source_mutex, and source.

Referenced by handle().

                                                       {
  std::shared_ptr<Source> source = nullptr;
  {
    std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
    if (m_activeSources.size() == 2) {
      if (m_nextInitialSourceToggle) {
        source = m_activeSources[0];
        m_nextInitialSourceToggle = false;
      } else {
        source = m_activeSources[1];
        m_nextInitialSourceToggle = true;
      }
    } else if (m_activeSources.empty()) {
      edm::Exception ex(edm::errors::FileReadError);
      ex << "XrdAdaptor::RequestManager::handle read(name='" << m_name << "', flags=0x" << std::hex << m_flags
         << ", permissions=0" << std::oct << m_perms << std::dec << ") => Source used after fatal exception.";
      ex.addContext("In XrdAdaptor::RequestManager::handle()");
      addConnections(ex);
      throw ex;
    } else {
      source = m_activeSources[0];
    }
  }
  return source;
}

prepareOpaqueString()

std::string RequestManager::prepareOpaqueString ( ) const

private

Prepare an opaque string appropriate for asking a redirector to open the current file but avoiding servers which we already have connections to.

Definition at line 639 of file XrdRequestManager.cc.

References submitPVResolutionJobs::count, ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, m_activeSources, m_disabledExcludeStrings, m_inactiveSources, m_source_mutex, contentValuesCheck::ss, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by initialize(), and XrdAdaptor::RequestManager::OpenHandler::open().

                                                    {
  struct {
    std::stringstream ss;
    size_t count = 0;
    bool has_active = false;

    void append_tried(const std::string &id, bool active = false) {
      ss << (count ? "," : "tried=") << id;
      count++;
      if (active) {
        has_active = true;
      }
    }
  } state;
  {
    std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);

    for (const auto &it : m_activeSources) {
      state.append_tried(it->ExcludeID().substr(0, it->ExcludeID().find(':')), true);
    }
    for (const auto &it : m_inactiveSources) {
      state.append_tried(it->ExcludeID().substr(0, it->ExcludeID().find(':')));
    }
  }
  for (const auto &it : m_disabledExcludeStrings) {
    state.append_tried(it.substr(0, it.find(':')));
  }
  if (state.has_active) {
    state.ss << "&triedrc=resel";
  }

  return state.ss.str();
}

queueUpdateCurrentServer()

void RequestManager::queueUpdateCurrentServer ( const std::string &  id )

private

Definition at line 314 of file XrdRequestManager.cc.

References XrdAdaptor::Source::getHostname(), l1ctLayer2EG_cff::id, m_serverToAdvertise, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by initialize().

                                                                 {
  auto hostname = std::make_unique<std::string>(id);
  if (Source::getHostname(id, *hostname)) {
    std::string *null_hostname = nullptr;
    if (m_serverToAdvertise.compare_exchange_strong(null_hostname, hostname.get())) {
      hostname.release();
    }
  }
}

reportSiteChange()

void RequestManager::reportSiteChange ( std::vector< std::shared_ptr< Source >> const &  iOld, std::vector< std::shared_ptr< Source >> const &  iNew, std::string  orig_site = std::string{}  ) const

private

Anytime we potentially switch sources, update the internal site source list; alert the user if necessary.

Definition at line 341 of file XrdRequestManager.cc.

References XrdAdaptor::Source::getHostname(), mps_fire::i, quality, AlCaHLTBitMon_QueryRunRegistry::string, and to_string().

Referenced by checkSourcesImpl(), compareSources(), initialize(), and requestFailure().

                                                                 {
  auto siteList = formatSites(iNew);
  if (orig_site.empty() || (orig_site == siteList)) {
    auto oldSites = formatSites(iOld);
  }

  edm::LogInfo("XrdAdaptorLvl1").log([&](auto &li) {
    li << "Serving data from: ";
    int size_active_sources = iNew.size();
    for (int i = 0; i < size_active_sources; ++i) {
      std::string hostname_a;
      Source::getHostname(iNew[i]->PrettyID(), hostname_a);
      li.format("   [{}] {}", i + 1, hostname_a);
    }
  });

  edm::LogInfo("XrdAdaptorLvl3").log([&](auto &li) {
    li << "The quality of the active sources is: ";
    int size_active_sources = iNew.size();
    for (int i = 0; i < size_active_sources; ++i) {
      std::string hostname_a;
      Source::getHostname(iNew[i]->PrettyID(), hostname_a);
      std::string quality = std::to_string(iNew[i]->getQuality());
      li.format("   [{}] {} for {}", i + 1, quality, hostname_a);
    }
  });
}

requestFailure()

void RequestManager::requestFailure	(	std::shared_ptr< XrdAdaptor::ClientRequest >	c_ptr,
		XrdCl::Status &	c_status
	)

Handle a failed client request.

Definition at line 818 of file XrdRequestManager.cc.

References cms::Exception::addAdditionalInfo(), addConnections(), cms::Exception::addContext(), TauDecayModes::dec, edm::errors::FileOpenError, edm::errors::FileReadError, spr::find(), GET_CLOCK_MONOTONIC, m_activeSources, m_disabledExcludeStrings, m_disabledSources, m_disabledSourceStrings, m_flags, m_lastSourceCheck, m_name, m_open_handler, m_perms, m_source_mutex, m_timeout, submitPVValidationJobs::now, reportSiteChange(), seconds(), mps_update::status, and mps_check::timeout.

                                                                                                       {
  std::shared_ptr<Source> source_ptr = c_ptr->getCurrentSource();

  // Fail early for invalid responses - XrdFile has a separate path for handling this.
  if (c_status.code == XrdCl::errInvalidResponse) {
    edm::LogWarning("XrdAdaptorInternal") << "Invalid response when reading from " << source_ptr->PrettyID();
    XrootdException ex(c_status, edm::errors::FileReadError);
    ex << "XrdAdaptor::RequestManager::requestFailure readv(name='" << m_name << "', flags=0x" << std::hex << m_flags
       << ", permissions=0" << std::oct << m_perms << std::dec << ", old source=" << source_ptr->PrettyID()
       << ") => Invalid ReadV response from server";
    ex.addContext("In XrdAdaptor::RequestManager::requestFailure()");
    addConnections(ex);
    throw ex;
  }
  edm::LogWarning("XrdAdaptorInternal") << "Request failure when reading from " << source_ptr->PrettyID();

  // Note that we do not delete the Source itself.  That is because this
  // function may be called from within XrdCl::ResponseHandler::HandleResponseWithHosts
  // In such a case, if you close a file in the handler, it will deadlock
  m_disabledSourceStrings.insert(source_ptr->ID());
  m_disabledExcludeStrings.insert(source_ptr->ExcludeID());
  m_disabledSources.insert(source_ptr);

  std::unique_lock<std::recursive_mutex> sentry(m_source_mutex);
  if ((!m_activeSources.empty()) && (m_activeSources[0].get() == source_ptr.get())) {
    auto oldSources = m_activeSources;
    m_activeSources.erase(m_activeSources.begin());
    reportSiteChange(oldSources, m_activeSources);
  } else if ((m_activeSources.size() > 1) && (m_activeSources[1].get() == source_ptr.get())) {
    auto oldSources = m_activeSources;
    m_activeSources.erase(m_activeSources.begin() + 1);
    reportSiteChange(oldSources, m_activeSources);
  }
  std::shared_ptr<Source> new_source;
  if (m_activeSources.empty()) {
    std::shared_future<std::shared_ptr<Source>> future = m_open_handler->open();
    timespec now;
    GET_CLOCK_MONOTONIC(now);
    m_lastSourceCheck = now;
    // Note we only wait for 180 seconds here.  This is because we've already failed
    // once and the likelihood the program has some inconsistent state is decent.
    // We'd much rather fail hard than deadlock!
    sentry.unlock();
    std::future_status status = future.wait_for(std::chrono::seconds(m_timeout + 10));
    if (status == std::future_status::timeout) {
      XrootdException ex(c_status, edm::errors::FileOpenError);
      ex << "XrdAdaptor::RequestManager::requestFailure Open(name='" << m_name << "', flags=0x" << std::hex << m_flags
         << ", permissions=0" << std::oct << m_perms << std::dec << ", old source=" << source_ptr->PrettyID()
         << ") => timeout when waiting for file open";
      ex.addContext("In XrdAdaptor::RequestManager::requestFailure()");
      addConnections(ex);
      throw ex;
    } else {
      try {
        new_source = future.get();
      } catch (edm::Exception &ex) {
        ex.addContext("Handling XrdAdaptor::RequestManager::requestFailure()");
        ex.addAdditionalInfo("Original failed source is " + source_ptr->PrettyID());
        throw;
      }
    }

    if (std::find(m_disabledSourceStrings.begin(), m_disabledSourceStrings.end(), new_source->ID()) !=
        m_disabledSourceStrings.end()) {
      // The server gave us back a data node we requested excluded.  Fatal!
      XrootdException ex(c_status, edm::errors::FileOpenError);
      ex << "XrdAdaptor::RequestManager::requestFailure Open(name='" << m_name << "', flags=0x" << std::hex << m_flags
         << ", permissions=0" << std::oct << m_perms << std::dec << ", old source=" << source_ptr->PrettyID()
         << ", new source=" << new_source->PrettyID() << ") => Xrootd server returned an excluded source";
      ex.addContext("In XrdAdaptor::RequestManager::requestFailure()");
      addConnections(ex);
      throw ex;
    }
    sentry.lock();

    auto oldSources = m_activeSources;
    m_activeSources.push_back(new_source);
    reportSiteChange(oldSources, m_activeSources);
  } else {
    new_source = m_activeSources[0];
  }
  new_source->handle(c_ptr);
}

splitClientRequest()

void XrdAdaptor::RequestManager::splitClientRequest ( const std::vector< IOPosBuffer > &  iolist, std::vector< IOPosBuffer > &  req1, std::vector< IOPosBuffer > &  req2, std::vector< std::shared_ptr< Source >> const &  activeSources  ) const

private

Given a client request, split it into two requests lists.

Definition at line 994 of file XrdRequestManager.cc.

References CalibrationSummaryClient_cfi::activeSources, cms::Exception::addAdditionalInfo(), cms::Exception::addContext(), cms::cuda::assert(), consumeChunkBack(), consumeChunkFront(), TauDecayModes::dec, edm::errors::FileReadError, ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, SiStripPI::max, edm::storage::IOPosBuffer::offset(), jetUpdater_cfi::sort, contentValuesCheck::ss, validateList(), XRD_ADAPTOR_CHUNK_THRESHOLD, and XRD_CL_MAX_CHUNK.

                                                                                                                 {
  if (iolist.empty())
    return;
  std::vector<IOPosBuffer> tmp_iolist(iolist.begin(), iolist.end());
  req1.reserve(iolist.size() / 2 + 1);
  req2.reserve(iolist.size() / 2 + 1);
  size_t front = 0;

  // The quality of both is increased by 5 to prevent strange effects if quality is 0 for one source.
  float q1 = static_cast<float>(activeSources[0]->getQuality()) + 5;
  float q2 = static_cast<float>(activeSources[1]->getQuality()) + 5;
  IOSize chunk1, chunk2;
  // Make sure the chunk size is at least 1024; little point to reads less than that size.
  chunk1 = std::max(static_cast<IOSize>(static_cast<float>(XRD_CL_MAX_CHUNK) * (q2 * q2 / (q1 * q1 + q2 * q2))),
                    static_cast<IOSize>(1024));
  chunk2 = std::max(static_cast<IOSize>(static_cast<float>(XRD_CL_MAX_CHUNK) * (q1 * q1 / (q1 * q1 + q2 * q2))),
                    static_cast<IOSize>(1024));

  IOSize size_orig = 0;
  for (const auto &it : iolist)
    size_orig += it.size();

  while (tmp_iolist.size() - front > 0) {
    if ((req1.size() >= XRD_ADAPTOR_CHUNK_THRESHOLD) &&
        (req2.size() >=
         XRD_ADAPTOR_CHUNK_THRESHOLD)) {  // The XrdFile::readv implementation should guarantee that no more than approximately 1024 chunks
      // are passed to the request manager.  However, because we have a max chunk size, we increase
      // the total number slightly.  Theoretically, it's possible an individual readv of total size >2GB where
      // each individual chunk is >1MB could result in this firing.  However, within the context of CMSSW,
      // this cannot happen (ROOT uses readv for TTreeCache; TTreeCache size is 20MB).
      edm::Exception ex(edm::errors::FileReadError);
      ex << "XrdAdaptor::RequestManager::splitClientRequest(name='" << m_name << "', flags=0x" << std::hex << m_flags
         << ", permissions=0" << std::oct << m_perms << std::dec
         << ") => Unable to split request between active servers.  This is an unexpected internal error and should be "
            "reported to CMSSW developers.";
      ex.addContext("In XrdAdaptor::RequestManager::requestFailure()");
      addConnections(ex);
      std::stringstream ss;
      ss << "Original request size " << iolist.size() << "(" << size_orig << " bytes)";
      ex.addAdditionalInfo(ss.str());
      std::stringstream ss2;
      ss2 << "Quality source 1 " << q1 - 5 << ", quality source 2: " << q2 - 5;
      ex.addAdditionalInfo(ss2.str());
      throw ex;
    }
    if (req1.size() < XRD_ADAPTOR_CHUNK_THRESHOLD) {
      consumeChunkFront(front, tmp_iolist, req1, chunk1);
    }
    if (req2.size() < XRD_ADAPTOR_CHUNK_THRESHOLD) {
      consumeChunkBack(front, tmp_iolist, req2, chunk2);
    }
  }
  std::sort(req1.begin(), req1.end(), [](const IOPosBuffer &left, const IOPosBuffer &right) {
    return left.offset() < right.offset();
  });
  std::sort(req2.begin(), req2.end(), [](const IOPosBuffer &left, const IOPosBuffer &right) {
    return left.offset() < right.offset();
  });

  IOSize size1 = validateList(req1);
  IOSize size2 = validateList(req2);

  assert(size_orig == size1 + size2);

  edm::LogVerbatim("XrdAdaptorInternal") << "Original request size " << iolist.size() << " (" << size_orig
                                         << " bytes) split into requests size " << req1.size() << " (" << size1
                                         << " bytes) and " << req2.size() << " (" << size2 << " bytes)" << std::endl;
}

updateCurrentServer()

void RequestManager::updateCurrentServer ( )

inlineprivate

Update the StatisticsSenderService, if necessary, with the current server.

Update the StatisticsSenderService with the current server info.

As this accesses the edm::Service infrastructure, this MUST be called from an edm-managed thread. It CANNOT be called from an Xrootd-managed thread.

Definition at line 297 of file XrdRequestManager.cc.

References edm::Service< T >::isAvailable(), LIKELY, m_name, m_serverToAdvertise, edm::storage::StatisticsSenderService::setCurrentServer(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by initialize().

                                         {
  // NOTE: we use memory_order_relaxed here, meaning that we may actually miss
  // a pending update.  *However*, since we call this for every read, we'll get it
  // eventually.
  if (LIKELY(!m_serverToAdvertise.load(std::memory_order_relaxed))) {
    return;
  }
  std::string *hostname_ptr;
  if ((hostname_ptr = m_serverToAdvertise.exchange(nullptr))) {
    std::unique_ptr<std::string> hostname(hostname_ptr);
    edm::Service<edm::storage::StatisticsSenderService> statsService;
    if (statsService.isAvailable()) {
      statsService->setCurrentServer(m_name, *hostname_ptr);
    }
  }
}

Member Data Documentation

m_activeSources

std::vector<std::shared_ptr<Source> > XrdAdaptor::RequestManager::m_activeSources

private

Note these member variables can only be accessed when the source mutex is held.

Definition at line 212 of file XrdRequestManager.h.

Referenced by getActiveFile(), getActiveSourceNames(), getPrettyActiveSourceNames(), handle(), initialize(), pickSingleSource(), prepareOpaqueString(), and requestFailure().

m_disabledExcludeStrings

oneapi::tbb::concurrent_unordered_set<std::string> XrdAdaptor::RequestManager::m_disabledExcludeStrings

private

Definition at line 216 of file XrdRequestManager.h.

Referenced by initialize(), prepareOpaqueString(), and requestFailure().

m_disabledSources

oneapi::tbb::concurrent_unordered_set<std::shared_ptr<Source>, SourceHash> XrdAdaptor::RequestManager::m_disabledSources

private

Definition at line 217 of file XrdRequestManager.h.

Referenced by requestFailure().

m_disabledSourceStrings

oneapi::tbb::concurrent_unordered_set<std::string> XrdAdaptor::RequestManager::m_disabledSourceStrings

private

Definition at line 215 of file XrdRequestManager.h.

Referenced by getDisabledSourceNames(), initialize(), and requestFailure().

m_distribution

std::uniform_real_distribution<float> XrdAdaptor::RequestManager::m_distribution

private

Definition at line 238 of file XrdRequestManager.h.

Referenced by checkSourcesImpl().

m_excluded_active_count

std::atomic<unsigned> XrdAdaptor::RequestManager::m_excluded_active_count

private

Definition at line 240 of file XrdRequestManager.h.

m_flags

XrdCl::OpenFlags::Flags XrdAdaptor::RequestManager::m_flags

private

Definition at line 233 of file XrdRequestManager.h.

Referenced by getActiveFile(), initialize(), XrdAdaptor::RequestManager::OpenHandler::open(), pickSingleSource(), and requestFailure().

m_generator

std::mt19937 XrdAdaptor::RequestManager::m_generator

private

Definition at line 237 of file XrdRequestManager.h.

Referenced by checkSourcesImpl().

m_inactiveSources

std::vector<std::shared_ptr<Source> > XrdAdaptor::RequestManager::m_inactiveSources

private

Definition at line 213 of file XrdRequestManager.h.

Referenced by handle(), and prepareOpaqueString().

m_lastSourceCheck

timespec XrdAdaptor::RequestManager::m_lastSourceCheck

private

Definition at line 223 of file XrdRequestManager.h.

Referenced by checkSources(), checkSourcesImpl(), initialize(), and requestFailure().

m_name

const std::string XrdAdaptor::RequestManager::m_name

private

Definition at line 232 of file XrdRequestManager.h.

Referenced by getActiveFile(), getFilename(), initialize(), XrdAdaptor::RequestManager::OpenHandler::open(), pickSingleSource(), requestFailure(), and updateCurrentServer().

m_nextActiveSourceCheck

timespec XrdAdaptor::RequestManager::m_nextActiveSourceCheck

private

Definition at line 228 of file XrdRequestManager.h.

Referenced by checkSources(), checkSourcesImpl(), and initialize().

m_nextInitialSourceToggle

bool XrdAdaptor::RequestManager::m_nextInitialSourceToggle

private

Definition at line 226 of file XrdRequestManager.h.

Referenced by pickSingleSource().

m_open_handler

std::shared_ptr<OpenHandler> XrdAdaptor::RequestManager::m_open_handler

private

Definition at line 299 of file XrdRequestManager.h.

Referenced by checkSourcesImpl(), initialize(), and requestFailure().

m_perms

XrdCl::Access::Mode XrdAdaptor::RequestManager::m_perms

private

Definition at line 234 of file XrdRequestManager.h.

Referenced by getActiveFile(), initialize(), XrdAdaptor::RequestManager::OpenHandler::open(), pickSingleSource(), and requestFailure().

m_redirectLimitDelayScale

int XrdAdaptor::RequestManager::m_redirectLimitDelayScale

private

Definition at line 229 of file XrdRequestManager.h.

m_serverToAdvertise

std::atomic<std::string *> XrdAdaptor::RequestManager::m_serverToAdvertise

private

Definition at line 221 of file XrdRequestManager.h.

Referenced by queueUpdateCurrentServer(), and updateCurrentServer().

m_source_mutex

std::recursive_mutex XrdAdaptor::RequestManager::m_source_mutex

mutableprivate

Definition at line 235 of file XrdRequestManager.h.

Referenced by getActiveFile(), getActiveSourceNames(), getPrettyActiveSourceNames(), handle(), initialize(), pickSingleSource(), prepareOpaqueString(), and requestFailure().

m_timeout

int XrdAdaptor::RequestManager::m_timeout

private

Definition at line 224 of file XrdRequestManager.h.

Referenced by initialize(), and requestFailure().

searchMode

bool XrdAdaptor::RequestManager::searchMode

private

Definition at line 230 of file XrdRequestManager.h.

XRD_DEFAULT_TIMEOUT

const unsigned int XrdAdaptor::RequestManager::XRD_DEFAULT_TIMEOUT = 3 * 60

static

Definition at line 53 of file XrdRequestManager.h.