XrdRequestManager.h

Go to the documentation of this file.

#ifndef Utilities_XrdAdaptor_XrdRequestManager_h
#define Utilities_XrdAdaptor_XrdRequestManager_h

#include <mutex>
#include <vector>
#include <set>
#include <condition_variable>
#include <random>
#include <sys/stat.h>

#include <boost/utility.hpp>
#include "tbb/concurrent_unordered_set.h"

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

#include "XrdCl/XrdClFileSystem.hh"

#include "XrdRequest.h"
#include "XrdSource.h"

namespace XrdCl {
    class File;
}


namespace XrdAdaptor {

struct SourceHash {
    using Key =std::shared_ptr<Source>;
    size_t operator()(const Key& iKey) const {
      return tbb::tbb_hasher(iKey.get());
    }
  };

  
class XrootdException : public edm::Exception {

public:

    XrootdException(XrdCl::Status & xrootd_status, edm::Exception::Code code)
      : Exception(code), m_code(xrootd_status.code)
    {}

    virtual ~XrootdException() throw() {};

    uint16_t getCode() { return m_code; }

private:

    uint16_t m_code;
};

class RequestManager : boost::noncopyable {

public:
    static const unsigned int XRD_DEFAULT_TIMEOUT = 3*60;

    ~RequestManager() = default;

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

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

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

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

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

    void addConnections(cms::Exception &) const;

    const std::string & getFilename() const {return m_name;}

    static std::shared_ptr<RequestManager>
    getInstance(const std::string &filename, XrdCl::OpenFlags::Flags flags, XrdCl::Access::Mode perms)
    {
        std::shared_ptr<RequestManager> instance(new RequestManager(filename, flags, perms));
        instance->initialize(instance);
        return instance;
    }

private:

    RequestManager(const std::string & filename, XrdCl::OpenFlags::Flags flags, XrdCl::Access::Mode perms);

    void initialize(std::weak_ptr<RequestManager> selfref);

    virtual void handleOpen(XrdCl::XRootDStatus &status, std::shared_ptr<Source>);

    void splitClientRequest(const std::vector<IOPosBuffer> &iolist,
                            std::vector<IOPosBuffer> &req1, std::vector<IOPosBuffer> &req2,
                            std::vector<std::shared_ptr<Source>> const& activeSources) const;

    void broadcastRequest(const ClientRequest &, bool active);

    void checkSources(timespec &now, IOSize requestSize,
                      std::vector<std::shared_ptr<Source>>& activeSources,
                      std::vector<std::shared_ptr<Source>>& inactiveSources); // TODO: inline
    void checkSourcesImpl(timespec &now, IOSize requestSize,
                          std::vector<std::shared_ptr<Source>>& activeSources,
                          std::vector<std::shared_ptr<Source>>& inactiveSources);
    bool compareSources(const timespec &now, unsigned a, unsigned b,
                        std::vector<std::shared_ptr<Source>>& activeSources,
                        std::vector<std::shared_ptr<Source>>& inactiveSources) const;

    void reportSiteChange(std::vector<std::shared_ptr<Source> > const& iOld,
                        std::vector<std::shared_ptr<Source> > const& iNew,
                        std::string orig_site=std::string{}) const;

    inline void updateCurrentServer();
    void queueUpdateCurrentServer(const std::string &);

    std::shared_ptr<Source> pickSingleSource();

    std::string prepareOpaqueString() const;

    std::vector<std::shared_ptr<Source> > m_activeSources;
    std::vector<std::shared_ptr<Source> > m_inactiveSources;
  
    tbb::concurrent_unordered_set<std::string> m_disabledSourceStrings;
    tbb::concurrent_unordered_set<std::string> m_disabledExcludeStrings;
    tbb::concurrent_unordered_set<std::shared_ptr<Source>, SourceHash> m_disabledSources;

    // StatisticsSenderService wants to know what our current server is;
    // this holds last-successfully-opened server name
    std::atomic<std::string*> m_serverToAdvertise;

    timespec m_lastSourceCheck;
    int m_timeout;
    // If set to true, the next active source should be 1; 0 otherwise.
    bool m_nextInitialSourceToggle;
    // The time when the next active source check should be performed.
    timespec m_nextActiveSourceCheck;
    bool searchMode;

    const std::string m_name;
    XrdCl::OpenFlags::Flags m_flags;
    XrdCl::Access::Mode m_perms;
    mutable std::recursive_mutex m_source_mutex;

    std::mt19937 m_generator;
    std::uniform_real_distribution<float> m_distribution;

    std::atomic<unsigned> m_excluded_active_count;

    class OpenHandler : boost::noncopyable, public XrdCl::ResponseHandler {

    public:

        static std::shared_ptr<OpenHandler> getInstance(std::weak_ptr<RequestManager> manager)
        {
            OpenHandler *instance_ptr = new OpenHandler(manager);
            std::shared_ptr<OpenHandler> instance(instance_ptr);
            instance_ptr->m_self_weak = instance;
            return instance;
        }

        ~OpenHandler();

        virtual void HandleResponseWithHosts(XrdCl::XRootDStatus *status, XrdCl::AnyObject *response, XrdCl::HostList *hostList) override;

        std::shared_future<std::shared_ptr<Source> > open();

        std::string current_source();

    private:

        OpenHandler(std::weak_ptr<RequestManager> manager);
        std::shared_future<std::shared_ptr<Source> > m_shared_future;
        std::promise<std::shared_ptr<Source> > m_promise;
        // Set to true only when there is an outstanding open request; not
        // protected by m_mutex, so the caller is required to know it is in a
        // thread-safe context.
        std::atomic<bool> m_outstanding_open {false};
        // Can only be touched when m_mutex is held.
        std::unique_ptr<XrdCl::File> m_file;
        std::recursive_mutex m_mutex;
        std::shared_ptr<OpenHandler> m_self;

        // Always maintain a weak self-reference; when the open is in-progress,
        // this is upgraded to a strong reference to prevent this object from
        // deletion as long as XrdCl has not performed the callback.
        std::weak_ptr<OpenHandler> m_self_weak;
        std::weak_ptr<RequestManager> m_manager;
    };

    std::shared_ptr<OpenHandler> m_open_handler;
};

}

#endif