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 "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 std::hash<Key::element_type *>{}(iKey.get()); }
  };
 
  class XrootdException : public edm::Exception {
  public:
    XrootdException(XrdCl::Status &xrootd_status, edm::Exception::Code code)
        : Exception(code), m_code(xrootd_status.code) {}
 
    ~XrootdException() noexcept override{};
 
    uint16_t getCode() { return m_code; }
 
  private:
    uint16_t m_code;
  };
 
  class RequestManager {
  public:
    RequestManager(const RequestManager &) = delete;
    RequestManager &operator=(const RequestManager &) = delete;
 
    static const unsigned int XRD_DEFAULT_TIMEOUT = 3 * 60;
 
    virtual ~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 : public XrdCl::ResponseHandler {
    public:
      OpenHandler(const OpenHandler &) = delete;
      OpenHandler &operator=(const OpenHandler &) = delete;
 
      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() override;
 
      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;
  };
 
}  // namespace XrdAdaptor
 
#endif