#include <XrdRequestManager.h>

Classes
class	OpenHandler

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 (std::shared_ptr< std::vector< IOPosBuffer >> iolist)

std::future< IOSize >	handle (std::shared_ptr< XrdAdaptor::ClientRequest > c_ptr)

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

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

tbb::concurrent_unordered_set< std::string >	m_disabledExcludeStrings

tbb::concurrent_unordered_set< std::shared_ptr< Source >, SourceHash >	m_disabledSources

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

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.

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 115 of file XrdRequestManager.cc.

     : m_serverToAdvertise(nullptr),
       m_timeout(XRD_DEFAULT_TIMEOUT),
       m_nextInitialSourceToggle(false),
       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 488 of file XrdRequestManager.cc.

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

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

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

◆ 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 300 of file XrdRequestManager.cc.

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

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

Referenced by handle().

◆ 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 346 of file XrdRequestManager.cc.

                                                                                          {
   bool findNewSource = false;
   if (activeSources.size() <= 1) {
     findNewSource = true;
   } 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;
 }

References CalibrationSummaryClient_cfi::activeSources, compareSources(), m_distribution, m_generator, m_lastSourceCheck, m_nextActiveSourceCheck, m_open_handler, eostools::move(), submitPVValidationJobs::now, alignCSCRings::r, reportSiteChange(), indexGen::s2, 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().

◆ 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 318 of file XrdRequestManager.cc.

                                                                                              {
   if (activeSources.size() < std::max(a, b) + 1) {
     return false;
   }
  
   bool findNewSource = false;
   if ((activeSources[a]->getQuality() > 5130) ||
       ((activeSources[a]->getQuality() > 260) &&
        (activeSources[b]->getQuality() * 4 < activeSources[a]->getQuality()))) {
     edm::LogVerbatim("XrdAdaptorInternal")
         << "Removing " << activeSources[a]->PrettyID() << " from active sources due to poor quality ("
         << activeSources[a]->getQuality() << " vs " << activeSources[b]->getQuality() << ")" << 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;
 }

References a, CalibrationSummaryClient_cfi::activeSources, b, SiStripPI::max, submitPVValidationJobs::now, and reportSiteChange().

Referenced by checkSources(), and checkSourcesImpl().

◆ getActiveFile()

std::shared_ptr< XrdCl::File > RequestManager::getActiveFile ( void ) const

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

Definition at line 452 of file XrdRequestManager.cc.

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

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

◆ 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 465 of file XrdRequestManager.cc.

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

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

◆ 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 481 of file XrdRequestManager.cc.

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

References m_disabledSourceStrings, source, and CalibrationSummaryClient_cfi::sources.

Referenced by addConnections().

◆ getFilename()

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

inline

Return current filename

Definition at line 103 of file XrdRequestManager.h.

103 { return m_name; }

References 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 112 of file XrdRequestManager.h.

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

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

◆ getPrettyActiveSourceNames()

void RequestManager::getPrettyActiveSourceNames ( std::vector< std::string > & sources ) const

Definition at line 473 of file XrdRequestManager.cc.

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

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

Referenced by addConnections().

◆ handle() [1/3]

std::future< IOSize > XrdAdaptor::RequestManager::handle ( std::shared_ptr< std::vector< IOPosBuffer >> iolist )

Definition at line 619 of file XrdRequestManager.cc.

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

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.

◆ handle() [2/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 527 of file XrdRequestManager.cc.

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

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.

◆ handle() [3/3]

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

inline

Interface for handling a client request.

Definition at line 56 of file XrdRequestManager.h.

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

References findQualityFiles::size.

◆ handleOpen()

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

privatevirtual

Handle the file-open response

Definition at line 581 of file XrdRequestManager.cc.

                                                                                                  {
   std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
   if (status.IsOK()) {
     edm::LogVerbatim("XrdAdaptorInternal") << "Successfully opened new source: " << source->PrettyID() << std::endl;
     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;
     m_nextActiveSourceCheck.tv_sec += XRD_ADAPTOR_LONG_OPEN_DELAY - XRD_ADAPTOR_SHORT_OPEN_DELAY;
   }
 }

References alignCSCRings::s, source, mps_update::status, XRD_ADAPTOR_LONG_OPEN_DELAY, and 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 125 of file XrdRequestManager.cc.

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

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, FrontierConditions_GlobalTag_cff::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.

◆ 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 501 of file XrdRequestManager.cc.

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

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().

◆ 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 554 of file XrdRequestManager.cc.

                                                     {
   std::stringstream ss;
   ss << "tried=";
   size_t count = 0;
   {
     std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
  
     for (const auto &it : m_activeSources) {
       count++;
       ss << it->ExcludeID().substr(0, it->ExcludeID().find(':')) << ",";
     }
     for (const auto &it : m_inactiveSources) {
       count++;
       ss << it->ExcludeID().substr(0, it->ExcludeID().find(':')) << ",";
     }
   }
   for (const auto &it : m_disabledExcludeStrings) {
     count++;
     ss << it.substr(0, it.find(':')) << ",";
   }
   if (count) {
     std::string tmp_str = ss.str();
     return tmp_str.substr(0, tmp_str.size() - 1);
   }
   return "";
 }

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

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

◆ queueUpdateCurrentServer()

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

private

Definition at line 258 of file XrdRequestManager.cc.

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

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

Referenced by initialize().

◆ 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 285 of file XrdRequestManager.cc.

                                                                  {
   auto siteList = formatSites(iNew);
   if (!orig_site.empty() && (orig_site != siteList)) {
     edm::LogWarning("XrdAdaptor") << "Data is served from " << siteList << " instead of original site " << orig_site;
   } else {
     auto oldSites = formatSites(iOld);
     if (orig_site.empty() && (siteList != oldSites)) {
       if (!oldSites.empty())
         edm::LogWarning("XrdAdaptor") << "Data is now served from " << siteList << " instead of previous " << oldSites;
     }
   }
 }

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

◆ requestFailure()

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

Handle a failed client request.

Definition at line 720 of file XrdRequestManager.cc.

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

References cms::Exception::addAdditionalInfo(), addConnections(), cms::Exception::addContext(), TauDecayModes::dec, edm::errors::FileOpenError, edm::errors::FileReadError, spr::find(), get, 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.

◆ 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 896 of file XrdRequestManager.cc.

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

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

◆ 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 241 of file XrdRequestManager.cc.

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

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

Referenced by initialize().

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 208 of file XrdRequestManager.h.

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

◆ m_disabledExcludeStrings

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

private

Definition at line 212 of file XrdRequestManager.h.

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

◆ m_disabledSources

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

private

Definition at line 213 of file XrdRequestManager.h.

Referenced by requestFailure().

◆ m_disabledSourceStrings

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

private

Definition at line 211 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 233 of file XrdRequestManager.h.

Referenced by checkSourcesImpl().

◆ m_excluded_active_count

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

private

Definition at line 235 of file XrdRequestManager.h.

◆ m_flags

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

private

Definition at line 228 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 232 of file XrdRequestManager.h.

Referenced by checkSourcesImpl().

◆ m_inactiveSources

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

private

Definition at line 209 of file XrdRequestManager.h.

Referenced by handle(), and prepareOpaqueString().

◆ m_lastSourceCheck

timespec XrdAdaptor::RequestManager::m_lastSourceCheck

private

Definition at line 219 of file XrdRequestManager.h.

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

◆ m_name

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

private

Definition at line 227 of file XrdRequestManager.h.

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

◆ m_nextActiveSourceCheck

timespec XrdAdaptor::RequestManager::m_nextActiveSourceCheck

private

Definition at line 224 of file XrdRequestManager.h.

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

◆ m_nextInitialSourceToggle

bool XrdAdaptor::RequestManager::m_nextInitialSourceToggle

private

Definition at line 222 of file XrdRequestManager.h.

Referenced by pickSingleSource().

◆ m_open_handler

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

private

Definition at line 294 of file XrdRequestManager.h.

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

◆ m_perms

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

private

Definition at line 229 of file XrdRequestManager.h.

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

◆ m_serverToAdvertise

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

private

Definition at line 217 of file XrdRequestManager.h.

Referenced by queueUpdateCurrentServer(), and updateCurrentServer().

◆ m_source_mutex

std::recursive_mutex XrdAdaptor::RequestManager::m_source_mutex

mutableprivate

Definition at line 230 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 220 of file XrdRequestManager.h.

Referenced by initialize(), and requestFailure().

◆ searchMode

bool XrdAdaptor::RequestManager::searchMode

private

Definition at line 225 of file XrdRequestManager.h.

◆ XRD_DEFAULT_TIMEOUT

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

static

Definition at line 49 of file XrdRequestManager.h.

Classes

Public Member Functions

Static Public Member Functions

Static Public Attributes

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ RequestManager() [1/2]

◆ ~RequestManager()

◆ RequestManager() [2/2]

Member Function Documentation

◆ addConnections()

◆ broadcastRequest()

◆ checkSources()

◆ checkSourcesImpl()

◆ compareSources()

◆ getActiveFile()

◆ getActiveSourceNames()

◆ getDisabledSourceNames()

◆ getFilename()

◆ getInstance()

◆ getPrettyActiveSourceNames()

◆ handle() [1/3]

◆ handle() [2/3]

◆ handle() [3/3]

◆ handleOpen()

◆ initialize()

◆ operator=()

◆ pickSingleSource()

◆ prepareOpaqueString()

◆ queueUpdateCurrentServer()

◆ reportSiteChange()

◆ requestFailure()

◆ splitClientRequest()

◆ updateCurrentServer()

Member Data Documentation

◆ m_activeSources

◆ m_disabledExcludeStrings

◆ m_disabledSources

◆ m_disabledSourceStrings

◆ m_distribution

◆ m_excluded_active_count

◆ m_flags

◆ m_generator

◆ m_inactiveSources

◆ m_lastSourceCheck

◆ m_name

◆ m_nextActiveSourceCheck

◆ m_nextInitialSourceToggle

◆ m_open_handler

◆ m_perms

◆ m_serverToAdvertise

◆ m_source_mutex

◆ m_timeout

◆ searchMode

◆ XRD_DEFAULT_TIMEOUT