#include <XrdRequestManager.h>

Inheritance diagram for XrdAdaptor::RequestManager:

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

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

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

Constructor & Destructor Documentation

XrdAdaptor::RequestManager::~RequestManager ( )

default

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

void RequestManager::addConnections ( cms::Exception & ex ) const

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

Definition at line 511 of file XrdRequestManager.cc.

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

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

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

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.

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

References checkSourcesImpl(), compareSources(), m_lastSourceCheck, m_nextActiveSourceCheck, 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);
     }
   }
 }

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

References compareSources(), m_distribution, m_generator, m_lastSourceCheck, m_nextActiveSourceCheck, m_open_handler, eostools::move(), cmsPerfSuiteHarvest::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().

 {
 
   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(std::move(*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;
 }

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

References a, b, hpstanc_transforms::max, and reportSiteChange().

Referenced by checkSources(), and checkSourcesImpl().

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

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

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

Definition at line 464 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();
 }

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

References m_activeSources, m_source_mutex, and source.

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

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

References m_disabledSourceStrings, and source.

Referenced by addConnections().

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

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

inline

Return current filename

Definition at line 111 of file XrdRequestManager.h.

111 {return m_name;}

XrdAdaptor::RequestManager::m_name

const std::string m_name

Definition: XrdRequestManager.h:231

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

References a, b, corrVsCorr::filename, flags, initialize(), instance, cmsPerfSuiteHarvest::now, mps_update::status, and AlCaHLTBitMon_QueryRunRegistry::string.

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

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

Definition at line 492 of file XrdRequestManager.cc.

References m_activeSources, m_source_mutex, and source.

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

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

inline

Interface for handling a client request.

Definition at line 63 of file XrdRequestManager.h.

References cmsBatch::handle, findQualityFiles::size, and btagGenBb_cfi::Status.

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

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

Definition at line 676 of file XrdRequestManager.cc.

References a, addConnections(), cms::Exception::addContext(), b, checkSources(), TauDecayModes::dec, edm::errors::FileReadError, GET_CLOCK_MONOTONIC, m_activeSources, m_flags, m_inactiveSources, m_name, m_perms, m_source_mutex, eostools::move(), cmsPerfSuiteHarvest::now, AlCaHLTBitMon_ParallelJobs::p, splitClientRequest(), edm::CPUTimer::start(), edm::CPUTimer::stop(), and updateCurrentServer().

 {
     //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->size())
     {
         c_ptr1.reset(new XrdAdaptor::ClientRequest(*this, req1));
         activeSources[0]->handle(c_ptr1);
         future1 = c_ptr1->get_future();
     }
     if (req2->size())
     {
         c_ptr2.reset(new XrdAdaptor::ClientRequest(*this, req2));
         activeSources[1]->handle(c_ptr2);
         future2 = c_ptr2->get_future();
     }
     if (req1->size() && req2->size())
     {
         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->size()) { return future1; }
     else if (req2->size()) { return future2; }
     else
     {   // Degenerate case - no bytes to read.
         std::promise<IOSize> p; p.set_value(0);
         return p.get_future();
     }
 }

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

References checkSources(), GET_CLOCK_MONOTONIC, m_activeSources, m_inactiveSources, m_source_mutex, eostools::move(), cmsPerfSuiteHarvest::now, pickSingleSource(), source, and AlCaHLTBitMon_QueryRunRegistry::string.

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

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

privatevirtual

Handle the file-open response

Definition at line 628 of file XrdRequestManager.cc.

References m_activeSources, m_excluded_active_count, m_inactiveSources, m_nextActiveSourceCheck, m_source_mutex, queueUpdateCurrentServer(), reportSiteChange(), alignCSCRings::s, 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;
         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;
     }
 }

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

References cms::Exception::addAdditionalInfo(), addConnections(), cms::Exception::addContext(), cms::Exception::clearAdditionalInfo(), cms::Exception::clearContext(), cms::Exception::clearMessage(), TauDecayModes::dec, XrdAdaptor::Source::determineHostExcludeString(), web.browse_db::env, FrontierConditions_GlobalTag_cff::file, ecalTB2006H4_GenSimDigiReco_cfg::File, edm::errors::FileOpenError, spr::find(), GET_CLOCK_MONOTONIC, XrdAdaptor::Source::getDomain(), XrdAdaptor::Source::getHostname(), XrdAdaptor::RequestManager::OpenHandler::getInstance(), XrdAdaptor::Source::getXrootdSiteFromURL(), training_settings::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.reset(new XrdCl::File());
     file->SetProperty("ReadRecovery", "false");
     auto opaque = prepareOpaqueString();
     std::string new_filename = m_name + (opaque.size() ? ((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.size())
       {
         ex.addAdditionalInfo("Problematic data server: " + dataServer);
       }
       if (lastUrl.size())
       {
         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.size())
       {
         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;
 }

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

private

Picks a single source for the next operation.

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

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

References KineDebug3::count(), m_activeSources, m_disabledExcludeStrings, m_inactiveSources, m_source_mutex, AlCaHLTBitMon_QueryRunRegistry::string, and create_public_lumi_plots::tmp_str.

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

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

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

private

Definition at line 279 of file XrdRequestManager.cc.

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

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

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

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

 {
   auto siteList = formatSites(iNew);
   if (orig_site.size() && (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.size() && (siteList != oldSites))
     {
       if (oldSites.size() >0 )
         edm::LogWarning("XrdAdaptor") << "Data is now served from " << siteList << " instead of previous " << oldSites;
     }
   }
 }

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

Handle a failed client request.

Definition at line 784 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, cmsPerfSuiteHarvest::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.size() > 0) && (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.size() == 0)
     {
         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);
 }

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

References cms::Exception::addAdditionalInfo(), addConnections(), cms::Exception::addContext(), consumeChunkBack(), consumeChunkFront(), TauDecayModes::dec, edm::errors::FileReadError, m_flags, m_name, m_perms, hpstanc_transforms::max, IOPosBuffer::offset(), q1, q2, validateList(), XRD_ADAPTOR_CHUNK_THRESHOLD, and XRD_CL_MAX_CHUNK.

Referenced by handle().

 {
     if (iolist.size() == 0) 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;
 }

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

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

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