#include <XrdRequestManager.h>

Inheritance diagram for XrdAdaptor::RequestManager:

Classes
class	OpenHandler

Public Member Functions
void	addConnections (cms::Exception &)

std::shared_ptr< XrdCl::File >	getActiveFile ()

void	getActiveSourceNames (std::vector< std::string > &sources)

void	getDisabledSourceNames (std::vector< std::string > &sources)

const std::string &	getFilename () const

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

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)

void	checkSourcesImpl (timespec &now, IOSize requestSize)

bool	compareSources (const timespec &now, unsigned a, unsigned b)

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

void	queueUpdateCurrentServer (const std::string &)

	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)

void	updateCurrentServer ()

void	updateSiteInfo (std::string orig_site="")

Private Attributes
std::string	m_activeSites

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

std::set< std::string >	m_disabledExcludeStrings

std::set< std::shared_ptr < Source > >	m_disabledSources

std::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 43 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 108 of file XrdRequestManager.cc.

Referenced by getInstance().

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

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

Definition at line 485 of file XrdRequestManager.cc.

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

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

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

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
	)

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

References checkSourcesImpl(), compareSources(), m_lastSourceCheck, m_nextActiveSourceCheck, m_source_mutex, 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.
       std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);   
       compareSources(now, 0, 1);
       compareSources(now, 1, 0);
     }
     if (timeDiffMS(now, m_nextActiveSourceCheck) > 0)
     {
       checkSourcesImpl(now, requestSize);
     }
   }
 }

void RequestManager::checkSourcesImpl	(	timespec &	now,
		IOSize	requestSize
	)

private

Definition at line 349 of file XrdRequestManager.cc.

References compareSources(), m_activeSources, m_distribution, m_generator, m_inactiveSources, m_lastSourceCheck, m_nextActiveSourceCheck, m_open_handler, m_source_mutex, eostools::move(), fileCollector::now, alignCSCRings::r, indexGen::s2, source, timeDiffMS(), updateSiteInfo(), XRD_ADAPTOR_LONG_OPEN_DELAY, XRD_ADAPTOR_OPEN_PROBE_PERCENT, XRD_ADAPTOR_SHORT_OPEN_DELAY, and XRD_ADAPTOR_SOURCE_QUALITY_FUDGE.

Referenced by checkSources().

 {
   std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
 
   bool findNewSource = false;
   if (m_activeSources.size() <= 1)
   {
     findNewSource = true;
   }
   else if (m_activeSources.size() > 1)
   {
     edm::LogVerbatim("XrdAdaptorInternal") << "Source 0 quality " << m_activeSources[0]->getQuality() << ", source 1 quality " << m_activeSources[1]->getQuality() << std::endl;
     findNewSource |= compareSources(now, 0, 1);
     findNewSource |= compareSources(now, 1, 0);
 
     // 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(m_inactiveSources.size());
     for (const auto & source : m_inactiveSources)
     {
       if (timeDiffMS(now, source->getLastDowngrade()) > (XRD_ADAPTOR_SHORT_OPEN_DELAY-1)*1000) {eligibleInactiveSources.push_back(source);}
     }
     std::vector<std::shared_ptr<Source> >::iterator 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();});
     std::vector<std::shared_ptr<Source> >::iterator worstActiveSource = std::max_element(m_activeSources.begin(), m_activeSources.end(),
         [](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()) && m_activeSources.size() == 1 && ((*bestInactiveSource)->getQuality() < 4*m_activeSources[0]->getQuality()))
     {
         m_activeSources.push_back(*bestInactiveSource);
         updateSiteInfo();
         for (auto it = m_inactiveSources.begin(); it != m_inactiveSources.end(); it++) if (it->get() == bestInactiveSource->get()) {m_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 = m_inactiveSources.begin(); it != m_inactiveSources.end(); it++) if (it->get() == bestInactiveSource->get()) {m_inactiveSources.erase(it); break;}
         m_inactiveSources.emplace_back(std::move(*worstActiveSource));
         m_activeSources.erase(worstActiveSource);
         m_activeSources.emplace_back(std::move(*bestInactiveSource));
         updateSiteInfo();
         eligibleInactiveSources.clear();
         for (const auto & source : m_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(m_activeSources.begin(), m_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 (m_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
	)

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

References a, b, m_activeSources, m_inactiveSources, bookConverter::max, and updateSiteInfo().

Referenced by checkSources(), and checkSourcesImpl().

 {
   if (m_activeSources.size() < std::max(a, b)+1) {return false;}
 
   bool findNewSource = false;
   if ((m_activeSources[a]->getQuality() > 5130) ||
      ((m_activeSources[a]->getQuality() > 260) && (m_activeSources[b]->getQuality()*4 < m_activeSources[a]->getQuality())))
   {
     edm::LogVerbatim("XrdAdaptorInternal") << "Removing "
           << m_activeSources[a]->PrettyID() << " from active sources due to poor quality ("
           << m_activeSources[a]->getQuality() << " vs " << m_activeSources[b]->getQuality() << ")" << std::endl;
     if (m_activeSources[a]->getLastDowngrade().tv_sec != 0) {findNewSource = true;}
     m_activeSources[a]->setLastDowngrade(now);
     m_inactiveSources.emplace_back(m_activeSources[a]);
     m_activeSources.erase(m_activeSources.begin()+a);
     updateSiteInfo();
   }
   return findNewSource;
 }

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

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

Definition at line 437 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 )

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

Definition at line 455 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 )

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

Definition at line 475 of file XrdRequestManager.cc.

References m_disabledSourceStrings, m_source_mutex, and source.

Referenced by addConnections().

 {
   std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
   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 101 of file XrdRequestManager.h.

References m_name.

101 {return m_name;}

XrdAdaptor::RequestManager::m_name

const std::string m_name

Definition: XrdRequestManager.h:210

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

References instance, and RequestManager().

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

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

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

     {
         std::shared_ptr<XrdAdaptor::ClientRequest> c_ptr(new 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 630 of file XrdRequestManager.cc.

References a, cms::Exception::addContext(), assert(), b, TauDecayModes::dec, edm::errors::FileReadError, GET_CLOCK_MONOTONIC, eostools::move(), fileCollector::now, AlCaHLTBitMon_ParallelJobs::p, edm::CPUTimer::start(), and edm::CPUTimer::stop().

 {
     std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
     updateCurrentServer();
 
     timespec now;
     GET_CLOCK_MONOTONIC(now);
 
     edm::CPUTimer timer;
     timer.start();
 
     if (m_activeSources.size() == 1)
     {
         std::shared_ptr<XrdAdaptor::ClientRequest> c_ptr(new XrdAdaptor::ClientRequest(*this, iolist));
         checkSources(now, c_ptr->getSize());
         m_activeSources[0]->handle(c_ptr);
         return c_ptr->get_future();
     }
     // Make sure active
     else if (m_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());
     std::shared_ptr<std::vector<IOPosBuffer> > req1(new std::vector<IOPosBuffer>);
     std::shared_ptr<std::vector<IOPosBuffer> > req2(new std::vector<IOPosBuffer>);
     splitClientRequest(*iolist, *req1, *req2);
 
     checkSources(now, req1->size() + req2->size());
     // CheckSources may have removed a source
     if (m_activeSources.size() == 1)
     {
         std::shared_ptr<XrdAdaptor::ClientRequest> c_ptr(new XrdAdaptor::ClientRequest(*this, iolist));
         m_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));
         m_activeSources[0]->handle(c_ptr1);
         future1 = c_ptr1->get_future();
     }
     if (req2->size())
     {
         c_ptr2.reset(new XrdAdaptor::ClientRequest(*this, req2));
         m_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 540 of file XrdRequestManager.cc.

References assert(), checkSources(), GET_CLOCK_MONOTONIC, fileCollector::now, pickSingleSource(), and source.

 {
   assert(c_ptr.get());
   timespec now;
   GET_CLOCK_MONOTONIC(now);
   checkSources(now, c_ptr->getSize());
 
   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 583 of file XrdRequestManager.cc.

References 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)
         {
             m_activeSources.push_back(source);
             updateSiteInfo();
             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 122 of file XrdRequestManager.cc.

References cms::Exception::addAdditionalInfo(), addConnections(), cms::Exception::addContext(), assert(), cms::Exception::clearAdditionalInfo(), cms::Exception::clearContext(), cms::Exception::clearMessage(), TauDecayModes::dec, XrdAdaptor::Source::determineHostExcludeString(), web.browse_db::env, mergeVDriftHistosByStation::file, reco_application_tbsim_DetSim-Digi_cfg::File, edm::errors::FileOpenError, spr::find(), GET_CLOCK_MONOTONIC, XrdAdaptor::Source::getDomain(), XrdAdaptor::Source::getHostname(), XrdAdaptor::RequestManager::OpenHandler::getInstance(), XrdAdaptor::Source::getXrootdSiteFromURL(), customizeTrackingMonitorSeedNumber::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(), prepareOpaqueString(), queueUpdateCurrentServer(), SendMonitoringInfo(), source, popcon_last_value_cfg::Source, ntuplemaker::status, AlCaHLTBitMon_QueryRunRegistry::string, updateCurrentServer(), updateSiteInfo(), 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());
     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);
 
   std::shared_ptr<Source> source(new Source(ts, std::move(file), excludeString));
   {
     std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
     m_activeSources.push_back(source);
     updateSiteInfo(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 502 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 ( )

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

References prof2calltree::count, m_activeSources, m_disabledExcludeStrings, m_inactiveSources, m_source_mutex, contentValuesCheck::ss, AlCaHLTBitMon_QueryRunRegistry::string, and create_public_lumi_plots::tmp_str.

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

 {
     std::lock_guard<std::recursive_mutex> sentry(m_source_mutex);
     std::stringstream ss;
     ss << "tried=";
     size_t count = 0;
     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 270 of file XrdRequestManager.cc.

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

Referenced by initialize().

 {
     std::unique_ptr<std::string> hostname(new 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::requestFailure	(	std::shared_ptr< XrdAdaptor::ClientRequest >	c_ptr,
		XrdCl::Status &	c_status
	)

Handle a failed client request.

Definition at line 723 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, fileCollector::now, seconds(), ntuplemaker::status, and updateSiteInfo().

 {
     std::unique_lock<std::recursive_mutex> sentry(m_source_mutex);
     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);
 
     if ((m_activeSources.size() > 0) && (m_activeSources[0].get() == source_ptr.get()))
     {
         m_activeSources.erase(m_activeSources.begin());
         updateSiteInfo();
     }
     else if ((m_activeSources.size() > 1) && (m_activeSources[1].get() == source_ptr.get()))
     {
         m_activeSources.erase(m_activeSources.begin()+1);
         updateSiteInfo();
     }
     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;
             }
         }
         sentry.lock();
         
         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;
         }
         m_activeSources.push_back(new_source);
         updateSiteInfo();
     }
     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
	)

private

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

Definition at line 938 of file XrdRequestManager.cc.

References cms::Exception::addAdditionalInfo(), cms::Exception::addContext(), assert(), consumeChunkBack(), consumeChunkFront(), TauDecayModes::dec, edm::errors::FileReadError, prof2calltree::front, bookConverter::max, IOPosBuffer::offset(), q1, q2, python.multivaluedict::sort(), contentValuesCheck::ss, validateList(), XRD_ADAPTOR_CHUNK_THRESHOLD, and XRD_CL_MAX_CHUNK.

 {
     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>(m_activeSources[0]->getQuality())+5;
     float q2 = static_cast<float>(m_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 251 of file XrdRequestManager.cc.

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

Referenced by initialize().

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

void RequestManager::updateSiteInfo ( std::string orig_site = "" )

private

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

Definition at line 284 of file XrdRequestManager.cc.

References m_activeSites, m_activeSources, and AlCaHLTBitMon_QueryRunRegistry::string.

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

 {
   std::string siteA, siteB, siteList;
   if (m_activeSources.size()) {siteA = m_activeSources[0]->Site();}
   if (m_activeSources.size() == 2) {siteB = m_activeSources[1]->Site();}
   siteList = siteA;
   if (siteB.size() && (siteB != siteA)) {siteList = siteA + ", " + siteB;}
   if (orig_site.size() && (orig_site != siteList))
   {
     edm::LogWarning("XrdAdaptor") << "Data is served from " << siteList << " instead of original site " << orig_site;
     m_activeSites = siteList;
   }
   else if (!orig_site.size() && (siteList != m_activeSites))
   {
     if (m_activeSites.size() >0 )
       edm::LogWarning("XrdAdaptor") << "Data is now served from " << siteList << " instead of previous " << m_activeSites;
     m_activeSites = siteList;
   }
 }