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

    ~XrootdException() noexcept override{};

    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;

    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 : 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() 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