XrdSource.cc

Go to the documentation of this file.

// See http://stackoverflow.com/questions/12523122/what-is-glibcxx-use-nanosleep-all-about
#define _GLIBCXX_USE_NANOSLEEP
#include <thread>
#include <chrono>
#include <atomic>
#include <iostream>
#include <cassert>
#include <netdb.h>

#include "XrdCl/XrdClFile.hh"

#include "FWCore/Utilities/interface/EDMException.h"
#include "FWCore/MessageLogger/interface/MessageLogger.h"

#include "XrdSource.h"
#include "XrdRequest.h"
#include "QualityMetric.h"
#include "XrdStatistics.h"

#define MAX_REQUEST 256*1024
#define XRD_CL_MAX_CHUNK 512*1024

#ifdef XRD_FAKE_SLOW
//#define XRD_DELAY 5140
#define XRD_DELAY 1000
#define XRD_SLOW_RATE 2
std::atomic<int> g_delayCount {0};
#else
std::atomic<int> g_delayCount {0};
#endif

using namespace XrdAdaptor;

// File::Close() can take awhile - slow servers (which are probably
// inactive anyway!) can even timeout.  Rather than wait around for
// a few minutes in the main thread, this class asynchronously closes
// and deletes the XrdCl::File
class DelayedClose : boost::noncopyable, public XrdCl::ResponseHandler
{
public:

    DelayedClose(std::shared_ptr<XrdCl::File> fh, const std::string &id, const std::string &site)
        : m_fh(std::move(fh)),
          m_id(id),
          m_site(site)
    {
        if (m_fh && m_fh->IsOpen())
        {
            if (!m_fh->Close(this).IsOK())
            {
                delete this;
            }
        }
    }


    ~DelayedClose() override = default;


    void HandleResponseWithHosts(XrdCl::XRootDStatus *status, XrdCl::AnyObject *response, XrdCl::HostList *hostList) override
    {
        if (status && !status->IsOK())
        {
            
            edm::LogWarning("XrdFileWarning") << "Source delayed close failed with error '" << status->ToStr()
                << "' (errno=" << status->errNo << ", code=" << status->code << ", server=" << m_id << ", site=" << m_site << ")";
        }
        delete status;
        delete hostList;
        // NOTE: we do not delete response (copying behavior from XrdCl).
        delete this;
    }


private:
    edm::propagate_const<std::shared_ptr<XrdCl::File>> m_fh;
    std::string m_id;
    std::string m_site;
};


class QueryAttrHandler : public XrdCl::ResponseHandler
{
    friend std::unique_ptr<QueryAttrHandler> std::make_unique<QueryAttrHandler>();

public:

    ~QueryAttrHandler() override = default;
    QueryAttrHandler(const QueryAttrHandler&) = delete;
    QueryAttrHandler& operator=(const QueryAttrHandler&) = delete;


    static XrdCl::XRootDStatus query(XrdCl::FileSystem &fs, const std::string &attr, std::chrono::milliseconds timeout, std::string &result)
    {
        auto handler = std::make_unique<QueryAttrHandler>();
        auto l_state = std::make_shared<QueryAttrState>();
        handler->m_state = l_state;
        XrdCl::Buffer arg(attr.size());
        arg.FromString(attr);

        XrdCl::XRootDStatus st = fs.Query(XrdCl::QueryCode::Config, arg, handler.get());
        if (!st.IsOK())
        {
            return st;
        }

        // Successfully registered the callback; it will always delete itself, so we shouldn't.
        handler.release();

        std::unique_lock<std::mutex> guard(l_state->m_mutex);
        // Wait until some status is available or a timeout.
        l_state->m_condvar.wait_for(guard, timeout, [&]{return l_state->m_status.get();});

        if (l_state->m_status)
        {
            if (l_state->m_status->IsOK())
            {
                result = l_state->m_response->ToString();
            }
            return *(l_state->m_status);
        }
        else
        {   // We had a timeout; construct a reasonable message.
            return XrdCl::XRootDStatus(XrdCl::stError, XrdCl::errSocketTimeout, 1, "Timeout when waiting for query callback.");
        }
    }


private:

    QueryAttrHandler() {}


    void HandleResponse(XrdCl::XRootDStatus *status, XrdCl::AnyObject *response ) override
    {
        // NOTE: we own the status and response pointers.
        std::unique_ptr<XrdCl::AnyObject> response_mgr;
        response_mgr.reset(response);

        // Lock our state information then dispose of our object.
        auto l_state = m_state.lock();
        delete this;
        if (!l_state) {return;}

        // On function exit, notify any waiting threads.
        std::unique_ptr<char, std::function<void(char*)>> notify_guard(nullptr, [&](char *) {l_state->m_condvar.notify_all();});

        {
            // On exit from the block, make sure m_status is set; it needs to be set before we notify threads.
            std::unique_ptr<char, std::function<void(char*)>> exit_guard(nullptr, [&](char *) {if (!l_state->m_status) l_state->m_status.reset(new XrdCl::XRootDStatus(XrdCl::stError, XrdCl::errInternal));});
            if (!status) {return;}
            if (status->IsOK())
            {
                if (!response) {return;}
                XrdCl::Buffer *buf_ptr;
                response->Get(buf_ptr);
                // AnyObject::Set lacks specialization for nullptr
                response->Set(static_cast<int *>(nullptr));
                l_state->m_response.reset(buf_ptr);
            }
            l_state->m_status.reset(status);
        }
    }


    // Represents the current state of the callback.  The parent class only manages a weak_ptr
    // to the state.  If the asynchronous callback cannot lock the weak_ptr, then it assumes the
    // main thread has given up and doesn't touch any of the state variables.
    struct QueryAttrState {

        // Synchronize between the callback thread and the main thread; condvar predicate
        // is having m_status set.  m_mutex protects m_status.
        std::mutex m_mutex;
        std::condition_variable m_condvar;

        // Results from the server
        std::unique_ptr<XrdCl::XRootDStatus> m_status;
        std::unique_ptr<XrdCl::Buffer> m_response;
    };
    std::weak_ptr<QueryAttrState> m_state;
};


Source::Source(timespec now, std::unique_ptr<XrdCl::File> fh, const std::string &exclude)
    : m_lastDowngrade({0, 0}),
      m_id("(unknown)"),
      m_exclude(exclude),
      m_fh(std::move(fh)),
      m_stats(nullptr)
#ifdef XRD_FAKE_SLOW
    , m_slow(++g_delayCount % XRD_SLOW_RATE == 0)
    //, m_slow(++g_delayCount >= XRD_SLOW_RATE)
    //, m_slow(true)
#endif
{
    if (m_fh.get())
    {
      if (!m_fh->GetProperty("DataServer", m_id))
      {
        edm::LogWarning("XrdFileWarning")
          << "Source::Source() failed to determine data server name.'";
      }
      if (m_exclude.empty()) {m_exclude = m_id;}
    }
    m_qm = QualityMetricFactory::get(now, m_id);
    m_prettyid = m_id + " (unknown site)";
    std::string domain_id;
    if (getDomain(m_id, domain_id)) {m_site = domain_id;}
    else {m_site = "Unknown (" + m_id + ")";}
    setXrootdSite();
    assert(m_qm.get());
    assert(m_fh.get());
    XrdSiteStatisticsInformation *statsService = XrdSiteStatisticsInformation::getInstance();
    if (statsService)
    {
        m_stats = statsService->getStatisticsForSite(m_site);
    }
}


bool Source::getHostname(const std::string &id, std::string &hostname)
{
    size_t pos = id.find(":");
    hostname = id;
    if ((pos != std::string::npos) && (pos > 0)) {hostname = id.substr(0, pos);}

    bool retval = true;
    if (!hostname.empty() && ((hostname[0] == '[') || isdigit(hostname[0])))
    {
        retval = false;
        struct addrinfo hints; memset(&hints, 0, sizeof(struct addrinfo));
        hints.ai_family = AF_UNSPEC;
        struct addrinfo *result;
        if (!getaddrinfo(hostname.c_str(), nullptr, &hints, &result))
        {
            std::vector<char> host; host.reserve(256);
            if (!getnameinfo(result->ai_addr, result->ai_addrlen, &host[0], 255, nullptr, 0, NI_NAMEREQD))
            {
                hostname = &host[0];
                retval = true;
            }
            freeaddrinfo(result);
        }
    }
    return retval;
}


bool Source::getDomain(const std::string &host, std::string &domain)
{
    getHostname(host, domain);
    size_t pos = domain.find(".");
    if (pos != std::string::npos && (pos < domain.size())) {domain = domain.substr(pos+1);}

    return !domain.empty();
}


bool
Source::isDCachePool(XrdCl::File &file, const XrdCl::HostList *hostList)
{
    // WORKAROUND: On open-file recovery in the Xrootd client, it'll carry around the
    // dCache opaque information to other sites, causing isDCachePool to erroneously return
    // true.  We are working with the upstream developers to solve this.
    //
    // For now, we see if the previous server also looks like a dCache pool - something that
    // wouldn't happen at a real site, as the previous server should look like a dCache door.
    std::string lastUrl;
    file.GetProperty("LastURL", lastUrl);
    if (!lastUrl.empty())
    {
        bool result = isDCachePool(lastUrl);
        if (result && hostList && (hostList->size() > 1))
        {
        if (isDCachePool((*hostList)[hostList->size()-2].url.GetURL()))
        {
            return false;
        }
        return true;
        }
    return result;
    }
    return false;
}

bool
Source::isDCachePool(const std::string &lastUrl)
{
    XrdCl::URL url(lastUrl);
    XrdCl::URL::ParamsMap map = url.GetParams();
    // dCache pools always utilize this opaque identifier.
    if (map.find("org.dcache.uuid") != map.end())
    {
        return true;
    }
    return false;
}


void
Source::determineHostExcludeString(XrdCl::File &file, const XrdCl::HostList *hostList, std::string &exclude)
{
    // Detect a dCache pool and, if we are in the federation context, give a custom
    // exclude parameter.
    // We assume this is a federation context if there's at least a regional, dCache door,
    // and dCache pool server (so, more than 2 servers!).

    exclude = "";
    if (hostList && (hostList->size() > 3) && isDCachePool(file, hostList))
    {
        const XrdCl::HostInfo &info = (*hostList)[hostList->size()-3];
        exclude = info.url.GetHostName();
        std::string lastUrl; file.GetProperty("LastURL", lastUrl);
        edm::LogVerbatim("XrdAdaptorInternal") << "Changing exclude list for URL " << lastUrl << " to " << exclude;
    }
}


bool
Source::getXrootdSite(XrdCl::File &fh, std::string &site)
{
    std::string lastUrl;
    fh.GetProperty("LastURL", lastUrl);
    if (lastUrl.empty() || isDCachePool(lastUrl))
    {
        std::string server, id;
        if (!fh.GetProperty("DataServer", server)) {id = "(unknown)";}
        else {id = server;}
        if (lastUrl.empty()) {edm::LogWarning("XrdFileWarning") << "Unable to determine the URL associated with server " << id;}
        site = "Unknown";
        if (!server.empty()) {getDomain(server, site);}
        return false;
    }
    return getXrootdSiteFromURL(lastUrl, site);
}

bool
Source::getXrootdSiteFromURL(std::string url, std::string &site)
{
    const std::string attr = "sitename";
    XrdCl::Buffer *response = nullptr;
    XrdCl::Buffer arg( attr.size() );
    arg.FromString( attr );

    XrdCl::FileSystem fs(url);
    std::string rsite;
    XrdCl::XRootDStatus st = QueryAttrHandler::query(fs, "sitename", std::chrono::seconds(1), rsite);
    if (!st.IsOK())
    {
        XrdCl::URL xurl(url);
        getDomain(xurl.GetHostName(), site);
        delete response;
        return false;
    }
    if (!rsite.empty() && (rsite[rsite.size()-1] == '\n'))
    {
        rsite = rsite.substr(0, rsite.size()-1);
    }
    if (rsite == "sitename")
    {
        XrdCl::URL xurl(url);
        getDomain(xurl.GetHostName(), site);
        return false;
    }
    site = rsite;
    return true;
}

void
Source::setXrootdSite()
{
    std::string site;
    bool goodSitename = getXrootdSite(*m_fh, site);
    if (!goodSitename)
    {
        edm::LogInfo("XrdAdaptorInternal")
          << "Xrootd server at " << m_id << " did not provide a sitename.  Monitoring may be incomplete.";
    }
    else
    {
       m_site = site;
        m_prettyid = m_id + " (site " + m_site + ")";
    }
    edm::LogInfo("XrdAdaptorInternal") << "Reading from new server " << m_id << " at site " << m_site;
}

Source::~Source()
{
  new DelayedClose(fh(), m_id, m_site);
}

std::shared_ptr<XrdCl::File>
Source::getFileHandle()
{
    return fh();
}

static void
validateList(const XrdCl::ChunkList& cl)
{
    off_t last_offset = -1;
    for (const auto & ci : cl)
    {
        assert(static_cast<off_t>(ci.offset) > last_offset);
        last_offset = ci.offset;
        assert(ci.length <= XRD_CL_MAX_CHUNK);
        assert(ci.offset < 0x1ffffffffff);
        assert(ci.offset > 0);
    }
    assert(cl.size() <= 1024);
}

void
Source::handle(std::shared_ptr<ClientRequest> c)
{
    edm::LogVerbatim("XrdAdaptorInternal") << "Reading from " << ID() << ", quality " << m_qm->get() << std::endl;
    c->m_source = shared_from_this();
    c->m_self_reference = c;
    m_qm->startWatch(c->m_qmw);
    if (m_stats)
    {
        std::shared_ptr<XrdReadStatistics> readStats = XrdSiteStatistics::startRead(stats(), c);
        c->setStatistics(readStats);
    }
#ifdef XRD_FAKE_SLOW
    if (m_slow) std::this_thread::sleep_for(std::chrono::milliseconds(XRD_DELAY));
#endif

    XrdCl::XRootDStatus status;
    if (c->m_into)
    {
        // See notes in ClientRequest definition to understand this voodoo.
        status = m_fh->Read(c->m_off, c->m_size, c->m_into, c.get());
    }
    else
    {
        XrdCl::ChunkList cl;
        cl.reserve(c->m_iolist->size());
        for (const auto & it : *c->m_iolist)
        {
            cl.emplace_back(it.offset(), it.size(), it.data());
        }
        validateList(cl);
        status = m_fh->VectorRead(cl, nullptr, c.get());
    }

    if (!status.IsOK())
    {
        edm::Exception ex(edm::errors::FileReadError);
        ex << "XrdFile::Read or XrdFile::VectorRead failed with error: '"
           << status.ToStr() << "' (errNo = " << status.errNo << ")";
        ex.addContext("Calling Source::handle");
        throw ex;
    }
}