XrdRequestManager.cc

Go to the documentation of this file.

#include <cassert>
#include <iostream>
#include <algorithm>
#include <netdb.h>

#include "XrdCl/XrdClFile.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClFileSystem.hh"

#include "FWCore/Utilities/interface/CPUTimer.h"
#include "FWCore/Utilities/interface/EDMException.h"
#include "FWCore/MessageLogger/interface/MessageLogger.h"
#include "FWCore/ServiceRegistry/interface/Service.h"
#include "FWCore/Utilities/interface/thread_safety_macros.h"
#include "Utilities/StorageFactory/interface/StatisticsSenderService.h"

#include "XrdStatistics.h"
#include "Utilities/XrdAdaptor/src/XrdRequestManager.h"
#include "Utilities/XrdAdaptor/src/XrdHostHandler.hh"

#define XRD_CL_MAX_CHUNK 512*1024

#define XRD_ADAPTOR_SHORT_OPEN_DELAY 5

#ifdef XRD_FAKE_OPEN_PROBE
#define XRD_ADAPTOR_OPEN_PROBE_PERCENT 100
#define XRD_ADAPTOR_LONG_OPEN_DELAY 20
// This is the minimal difference in quality required to swap an active and inactive source
#define XRD_ADAPTOR_SOURCE_QUALITY_FUDGE 0
#else
#define XRD_ADAPTOR_OPEN_PROBE_PERCENT 10
#define XRD_ADAPTOR_LONG_OPEN_DELAY 2*60
#define XRD_ADAPTOR_SOURCE_QUALITY_FUDGE 100
#endif

#define XRD_ADAPTOR_CHUNK_THRESHOLD 1000


#ifdef __MACH__
#include <mach/clock.h>
#include <mach/mach.h>
#define GET_CLOCK_MONOTONIC(ts) \
{ \
  clock_serv_t cclock; \
  mach_timespec_t mts; \
  host_get_clock_service(mach_host_self(), SYSTEM_CLOCK, &cclock); \
  clock_get_time(cclock, &mts); \
  mach_port_deallocate(mach_task_self(), cclock); \
  ts.tv_sec = mts.tv_sec; \
  ts.tv_nsec = mts.tv_nsec; \
}
#else
#define GET_CLOCK_MONOTONIC(ts) \
  clock_gettime(CLOCK_MONOTONIC, &ts);
#endif

using namespace XrdAdaptor;

long long timeDiffMS(const timespec &a, const timespec &b)
{
  long long diff = (a.tv_sec - b.tv_sec) * 1000;
  diff += (a.tv_nsec - b.tv_nsec) / 1e6;
  return diff;
}

/*
 * We do not care about the response of sending the monitoring information;
 * this handler class simply frees any returned buffer to prevent memory leaks.
 */
class SendMonitoringInfoHandler : boost::noncopyable, public XrdCl::ResponseHandler
{
    void HandleResponse(XrdCl::XRootDStatus *status, XrdCl::AnyObject *response) override
    {
        if (response)
        {
            XrdCl::Buffer *buffer = nullptr;
            response->Get(buffer);
            response->Set(static_cast<int*>(nullptr));
            delete buffer;
        }
        // Send Info has a response object; we must delete it.
        delete response;
        delete status;
    }
};

CMS_THREAD_SAFE SendMonitoringInfoHandler nullHandler;


static void
SendMonitoringInfo(XrdCl::File &file)
{
        // Do not send this to a dCache data server as they return an error.
        // In some versions of dCache, sending the monitoring information causes
        // the server to close the connection - resulting in failures.
    if (Source::isDCachePool(file)) {return;}

    // Send the monitoring info, if available.
    const char * jobId = edm::storage::StatisticsSenderService::getJobID();
    std::string lastUrl;
    file.GetProperty("LastURL", lastUrl);
    if (jobId && !lastUrl.empty())
    {
        XrdCl::URL url(lastUrl);
        XrdCl::FileSystem fs(url);
        if (!(fs.SendInfo(jobId, &nullHandler, 30).IsOK()))
        {
            edm::LogWarning("XrdAdaptorInternal") << "Failed to send the monitoring information, monitoring ID is " << jobId << ".";
        }
        edm::LogInfo("XrdAdaptorInternal") << "Set monitoring ID to " << jobId << ".";
    }
}


RequestManager::RequestManager(const std::string &filename, XrdCl::OpenFlags::Flags flags, XrdCl::Access::Mode perms)
    : 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)
{
}


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

  timespec ts;
  GET_CLOCK_MONOTONIC(ts);

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

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

void
RequestManager::updateCurrentServer()
{
    // 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::queueUpdateCurrentServer(const std::string &id)
{
    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();
        }
    }
}

namespace  {
  std::string formatSites(std::vector<std::shared_ptr<Source> > const& iSources) {
    std::string siteA, siteB;
    if (!iSources.empty()) {siteA = iSources[0]->Site();}
    if (iSources.size() == 2) {siteB = iSources[1]->Site();}
    std::string siteList = siteA;
    if (!siteB.empty() && (siteB != siteA)) {siteList = siteA + ", " + siteB;}
    return siteList;
  }
}
  
void
RequestManager::reportSiteChange(std::vector<std::shared_ptr<Source> > const& iOld,
                               std::vector<std::shared_ptr<Source> > const& iNew,
                               std::string orig_site) const
{
  auto siteList = formatSites(iNew);
  if (!orig_site.empty() && (orig_site != siteList))
  {
    edm::LogWarning("XrdAdaptor") << "Data is served from " << siteList << " instead of original site " << orig_site;
  }
  else {
    auto oldSites = formatSites(iOld);
    if (orig_site.empty() && (siteList != oldSites))
    {
      if (!oldSites.empty() )
        edm::LogWarning("XrdAdaptor") << "Data is now served from " << siteList << " instead of previous " << oldSites;
    }
  }
}


void
RequestManager::checkSources(timespec &now, IOSize requestSize,
                             std::vector<std::shared_ptr<Source>>& activeSources,
                             std::vector<std::shared_ptr<Source>>& inactiveSources)
{
  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);
    }
  }
}


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

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

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

std::shared_ptr<XrdCl::File>
RequestManager::getActiveFile() const
{
  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
{
  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::getPrettyActiveSourceNames(std::vector<std::string> & sources) const
{
  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());
  }
}

void
RequestManager::getDisabledSourceNames(std::vector<std::string> & sources) const
{
  sources.reserve(m_disabledSourceStrings.size());
  for (auto const& source : m_disabledSourceStrings) {
    sources.push_back(source);
  }
}

void
RequestManager::addConnections(cms::Exception &ex) const
{
  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);
  }
}

std::shared_ptr<Source>
RequestManager::pickSingleSource()
{
  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::future<IOSize>
RequestManager::handle(std::shared_ptr<XrdAdaptor::ClientRequest> c_ptr)
{
  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();
}

std::string
RequestManager::prepareOpaqueString() const
{
    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 
XrdAdaptor::RequestManager::handleOpen(XrdCl::XRootDStatus &status, std::shared_ptr<Source> source)
{
    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;
    }
}

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

    timespec now;
    GET_CLOCK_MONOTONIC(now);

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

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

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

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

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

void
RequestManager::requestFailure(std::shared_ptr<XrdAdaptor::ClientRequest> c_ptr, XrdCl::Status &c_status)
{
    std::shared_ptr<Source> source_ptr = c_ptr->getCurrentSource();

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

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

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

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

static void
consumeChunkFront(size_t &front, std::vector<IOPosBuffer> &input, std::vector<IOPosBuffer> &output, IOSize chunksize)
{
    while ((chunksize > 0) && (front < input.size()) && (output.size() <= XRD_ADAPTOR_CHUNK_THRESHOLD))
    {
        IOPosBuffer &io = input[front];
        IOPosBuffer &outio = output.back();
        if (io.size() > chunksize)
        {
            IOSize consumed;
            if (!output.empty() && (outio.size() < XRD_CL_MAX_CHUNK) && (outio.offset() + static_cast<IOOffset>(outio.size()) == io.offset()))
            {
                if (outio.size() + chunksize > XRD_CL_MAX_CHUNK)
                {
                    consumed = (XRD_CL_MAX_CHUNK - outio.size());
                    outio.set_size(XRD_CL_MAX_CHUNK);
                }
                else
                {
                    consumed = chunksize;
                    outio.set_size(outio.size() + consumed);
                }
            }
            else
            {
                consumed = chunksize;
                output.emplace_back(IOPosBuffer(io.offset(), io.data(), chunksize));
            }
            chunksize -= consumed;
            IOSize newsize = io.size() - consumed;
            IOOffset newoffset = io.offset() + consumed;
            void* newdata = static_cast<char*>(io.data()) + consumed;
            io.set_offset(newoffset);
            io.set_data(newdata);
            io.set_size(newsize);
        }
        else if (io.size() == 0)
        {
            front++;
        }
        else
        {
            output.push_back(io);
            chunksize -= io.size();
            front++;
        }
    }
}

static void
consumeChunkBack(size_t front, std::vector<IOPosBuffer> &input, std::vector<IOPosBuffer> &output, IOSize chunksize)
{
    while ((chunksize > 0) && (front < input.size()) && (output.size() <= XRD_ADAPTOR_CHUNK_THRESHOLD))
    {
        IOPosBuffer &io = input.back();
        IOPosBuffer &outio = output.back();
        if (io.size() > chunksize)
        {
            IOSize consumed;
            if (!output.empty() && (outio.size() < XRD_CL_MAX_CHUNK) && (outio.offset() + static_cast<IOOffset>(outio.size()) == io.offset()))
            {
                if (outio.size() + chunksize > XRD_CL_MAX_CHUNK)
                {
                    consumed = (XRD_CL_MAX_CHUNK - outio.size());
                    outio.set_size(XRD_CL_MAX_CHUNK);
                }
                else
                {
                    consumed = chunksize;
                    outio.set_size(outio.size() + consumed);
                }
            }
            else
            {
                consumed = chunksize;
                output.emplace_back(IOPosBuffer(io.offset(), io.data(), chunksize));
            }
            chunksize -= consumed;
            IOSize newsize = io.size() - consumed;
            IOOffset newoffset = io.offset() + consumed;
            void* newdata = static_cast<char*>(io.data()) + consumed;
            io.set_offset(newoffset);
            io.set_data(newdata);
            io.set_size(newsize);
        }
        else if (io.size() == 0)
        {
            input.pop_back();
        }
        else
        {
            output.push_back(io);
            chunksize -= io.size();
            input.pop_back();
        }
    }
}

static IOSize validateList(const std::vector<IOPosBuffer> req)
{
    IOSize total = 0;
    off_t last_offset = -1;
    for (const auto & it : req)
    {
        total += it.size();
        assert(it.offset() > last_offset);
        last_offset = it.offset();
        assert(it.size() <= XRD_CL_MAX_CHUNK);
        assert(it.offset() < 0x1ffffffffff);
    }
    assert(req.size() <= 1024);
    return total;
}

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

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

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

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

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

    assert(size_orig == size1 + size2);

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

XrdAdaptor::RequestManager::OpenHandler::OpenHandler(std::weak_ptr<RequestManager> manager)
  : m_manager(manager)
{
}


    // Cannot use ~OpenHandler=default as XrdCl::File is not fully
    // defined in the header.
XrdAdaptor::RequestManager::OpenHandler::~OpenHandler()
{
}


void
XrdAdaptor::RequestManager::OpenHandler::HandleResponseWithHosts(XrdCl::XRootDStatus *status_ptr, XrdCl::AnyObject *, XrdCl::HostList *hostList_ptr)
{
  // Make sure we get rid of the strong self-reference when the callback finishes.
  std::shared_ptr<OpenHandler> self = m_self;
  m_self.reset();

  // NOTE: as in XrdCl::File (synchronous), we ignore the response object.
  // Make sure that we set m_outstanding_open to false on exit from this function.
  // NOTE: we need to pass non-nullptr to unique_ptr in order for the guard to run
  std::unique_ptr<OpenHandler, std::function<void(OpenHandler*)>> outstanding_guard(this, [&](OpenHandler*){m_outstanding_open=false;});

  std::shared_ptr<Source> source;
  std::unique_ptr<XrdCl::XRootDStatus> status(status_ptr);
  std::unique_ptr<XrdCl::HostList> hostList(hostList_ptr);

  auto manager = m_manager.lock();
    // Manager object has already been deleted.  Cleanup the
    // response objects, remove our self-reference, and ignore the response.
  if (!manager)
  {
    return;
  }
  //if we need to delete the File object we must do it outside
  // of the lock to avoid a potential deadlock
  std::unique_ptr<XrdCl::File> releaseFile;
  {
    std::lock_guard<std::recursive_mutex> sentry(m_mutex);

    if (status->IsOK())
    {
        SendMonitoringInfo(*m_file);
        timespec now;
        GET_CLOCK_MONOTONIC(now);

        std::string excludeString;
        Source::determineHostExcludeString(*m_file, hostList.get(), excludeString);

        source.reset(new Source(now, std::move(m_file), excludeString));
        m_promise.set_value(source);
    }
    else
    {
        releaseFile = std::move(m_file);
        edm::Exception ex(edm::errors::FileOpenError);
        ex << "XrdCl::File::Open(name='" << manager->m_name
           << "', flags=0x" << std::hex << manager->m_flags
           << ", permissions=0" << std::oct << manager->m_perms << std::dec
           << ") => error '" << status->ToStr()
           << "' (errno=" << status->errNo << ", code=" << status->code << ")";
        ex.addContext("In XrdAdaptor::RequestManager::OpenHandler::HandleResponseWithHosts()");
        manager->addConnections(ex);

        m_promise.set_exception(std::make_exception_ptr(ex));
    }
  }
  manager->handleOpen(*status, source);
}

std::string
XrdAdaptor::RequestManager::OpenHandler::current_source()
{
    std::lock_guard<std::recursive_mutex> sentry(m_mutex);

    if (!m_file.get())
    {
        return "(no open in progress)";
    }
    std::string dataServer;
    m_file->GetProperty("DataServer", dataServer);
    if (dataServer.empty()) { return "(unknown source)"; }
    return dataServer;
}

std::shared_future<std::shared_ptr<Source> >
XrdAdaptor::RequestManager::OpenHandler::open()
{
    auto manager_ptr = m_manager.lock();
    if (!manager_ptr)
    {
      edm::Exception ex(edm::errors::LogicError);
      ex << "XrdCl::File::Open() =>"
         << " error: OpenHandler called within an invalid RequestManager context."
         << "  This is a logic error and should be reported to the CMSSW developers.";
      ex.addContext("Calling XrdAdaptor::RequestManager::OpenHandler::open()");
      throw ex;
    }
    RequestManager &manager = *manager_ptr;
    auto self_ptr = m_self_weak.lock();
    if (!self_ptr)
    {
      edm::Exception ex(edm::errors::LogicError);
      ex << "XrdCl::File::Open() => error: "
         << "OpenHandler called after it was deleted.  This is a logic error "
         << "and should be reported to the CMSSW developers.";
      ex.addContext("Calling XrdAdapter::RequestManager::OpenHandler::open()");
      throw ex;
    }

      // NOTE NOTE: we look at this variable *without* the lock.  This means the method
      // is not thread-safe; the caller is responsible to verify it is not called from
      // multiple threads simultaneously.
      //
      // This is done because ::open may be called from a Xrootd callback; if we
      // tried to hold m_mutex here, this object's callback may also be active, hold m_mutex,
      // and make a call into xrootd (when it invokes m_file.reset()).  Hence, our callback
      // holds our mutex and attempts to grab an Xrootd mutex; RequestManager::requestFailure holds
      // an Xrootd mutex and tries to hold m_mutex.  This is a classic deadlock.
    if (m_outstanding_open)
    {
        return m_shared_future;
    }
    std::lock_guard<std::recursive_mutex> sentry(m_mutex);
    std::promise<std::shared_ptr<Source> > new_promise;
    m_promise.swap(new_promise);
    m_shared_future = m_promise.get_future().share();

    auto opaque = manager.prepareOpaqueString();
    std::string new_name = manager.m_name + ((manager.m_name.find("?") == manager.m_name.npos) ? "?" : "&") + opaque;
    edm::LogVerbatim("XrdAdaptorInternal") << "Trying to open URL: " << new_name;
    m_file.reset(new XrdCl::File());
    m_outstanding_open = true;

    // Always make sure we release m_file and set m_outstanding_open to false on error.
    std::unique_ptr<OpenHandler, std::function<void(OpenHandler*)>> exit_guard(this, [&](OpenHandler*){m_outstanding_open = false; m_file.reset();});

    XrdCl::XRootDStatus status;
    if (!(status = m_file->Open(new_name, manager.m_flags, manager.m_perms, this)).IsOK())
    {
      edm::Exception ex(edm::errors::FileOpenError);
      ex << "XrdCl::File::Open(name='" << new_name
         << "', flags=0x" << std::hex << manager.m_flags
         << ", permissions=0" << std::oct << manager.m_perms << std::dec
         << ") => error '" << status.ToStr()
         << "' (errno=" << status.errNo << ", code=" << status.code << ")";
      ex.addContext("Calling XrdAdaptor::RequestManager::OpenHandler::open()");
      manager.addConnections(ex);
      throw ex;
    }
    exit_guard.release();
      // Have a strong self-reference for as long as the callback is in-progress.
    m_self = self_ptr;
    return m_shared_future;
}