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 "FWCore/Utilities/interface/EDMException.h"

#include "XrdCl/XrdClFileSystem.hh"

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

namespace XrdCl {
    class File;
}

namespace XrdAdaptor {

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)
    {
        std::shared_ptr<XrdAdaptor::ClientRequest> c_ptr(new 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);
    void getPrettyActiveSourceNames(std::vector<std::string> & sources);

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

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

    void addConnections(cms::Exception &);

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

    void broadcastRequest(const ClientRequest &, bool active);

    void checkSources(timespec &now, IOSize requestSize); // TODO: inline
    void checkSourcesImpl(timespec &now, IOSize requestSize);
    bool compareSources(const timespec &now, unsigned a, unsigned b);

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

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

    std::shared_ptr<Source> pickSingleSource();

    std::string prepareOpaqueString();

    std::vector<std::shared_ptr<Source> > m_activeSources;
    std::vector<std::shared_ptr<Source> > m_inactiveSources;
    std::set<std::string> m_disabledSourceStrings;
    std::set<std::string> m_disabledExcludeStrings;
    std::set<std::shared_ptr<Source> > m_disabledSources;
    std::string m_activeSites;
    // 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;
    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;
        // When this is not null, there is a file-open in process
        // 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