Merge remote-tracking branch 'origin/dev' into stable-sync

Author: davidrohr

Algorithm/test/pageparser.cxx

@@ -50,7 +50,7 @@ struct ClusterData {
   {
   }
 
-  bool operator==(const ClusterData& rhs)
+  bool operator==(const ClusterData& rhs) const
   {
     return clusterid == rhs.clusterid && x == rhs.x && y == rhs.y && z == rhs.z && e == rhs.e;
   }

CCDB/include/CCDB/CCDBDownloader.h

@@ -11,6 +11,10 @@
 #ifndef O2_CCDBDOWNLOADER_H_
 #define O2_CCDBDOWNLOADER_H_
 
+#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
+#include "MemoryResources/MemoryResources.h"
+#endif
+
 #include <cstdio>
 #include <cstdlib>
 #include <curl/curl.h>
@@ -21,6 +25,8 @@
 #include <mutex>
 #include <condition_variable>
 #include <unordered_map>
+#include <map>
+#include <functional>
 
 typedef struct uv_loop_s uv_loop_t;
 typedef struct uv_timer_s uv_timer_t;
@@ -32,6 +38,25 @@ typedef struct uv_handle_s uv_handle_t;
 namespace o2::ccdb
 {
 
+#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
+struct HeaderObjectPair_t {
+  std::multimap<std::string, std::string> header;
+  o2::pmr::vector<char>* object = nullptr;
+  int counter = 0;
+};
+
+typedef struct DownloaderRequestData {
+  std::vector<std::string> hosts;
+  std::string path;
+  long timestamp;
+  HeaderObjectPair_t hoPair;
+  std::map<std::string, std::string>* headers;
+
+  std::function<bool(std::string)> localContentCallback;
+  bool errorflag = false;
+} DownloaderRequestData;
+#endif
+
 /*
  Some functions below aren't member functions of CCDBDownloader because both curl and libuv require callback functions which have to be either static or non-member.
  Because non-static functions are used in the functions below, they must be non-member.
@@ -133,6 +158,14 @@ class CCDBDownloader
    */
   std::vector<CURLcode> batchBlockingPerform(std::vector<CURL*> const& handleVector);
 
+  /**
+   * Schedules an asynchronous transfer but doesn't perform it.
+   *
+   * @param handle Handle to be performed on.
+   * @param requestCounter Counter shared by a batch of CURL handles.
+   */
+  void asynchSchedule(CURL* handle, size_t* requestCounter);
+
   /**
    * Limits the number of parallel connections. Should be used only if no transfers are happening.
    */
@@ -176,6 +209,24 @@ class CCDBDownloader
   void runLoop(bool noWait);
 
  private:
+  /**
+   * Leaves only the protocol and host part of the url, discrading path and metadata.
+   */
+  std::string trimHostUrl(std::string full_host_url) const;
+
+  /**
+   * Recognizes whether the address is a full url, or a partial one (like for example "/Task/Detector/1") and combines it with potentialHost if needed.
+   */
+  std::string prepareRedirectedURL(std::string address, std::string potentialHost) const;
+
+  /**
+   * Returns a vector of possible content locations based on the redirect headers.
+   *
+   * @param baseUrl Content path.
+   * @param headerMap Map containing response headers.
+   */
+  std::vector<std::string> getLocations(std::multimap<std::string, std::string>* headerMap) const;
+
   std::string mUserAgentId = "CCDBDownloader";
   /**
    * Sets up internal UV loop.
@@ -207,8 +258,7 @@ class CCDBDownloader
    */
   enum RequestType {
     BLOCKING,
-    ASYNCHRONOUS,
-    ASYNCHRONOUS_WITH_CALLBACK
+    ASYNCHRONOUS
   };
 
   /**
@@ -230,20 +280,19 @@ class CCDBDownloader
 
   DataForSocket mSocketData;
 
+#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
   /**
    * Structure which is stored in a easy_handle. It carries information about the request which the easy_handle is part of.
-   * All easy handles coming from one request have an identical PerformData structure.
    */
   typedef struct PerformData {
-    std::condition_variable* cv;
-    bool* completionFlag;
     CURLcode* codeDestination;
-    void (*cbFun)(void*);
-    std::thread* cbThread;
-    void* cbData;
     size_t* requestsLeft;
     RequestType type;
+    int hostInd;
+    int locInd;
+    DownloaderRequestData* requestData;
   } PerformData;
+#endif
 
   /**
    * Called by CURL in order to close a socket. It will be called by CURL even if a timeout timer closed the socket beforehand.
@@ -253,6 +302,23 @@ class CCDBDownloader
    */
   static void closesocketCallback(void* clientp, curl_socket_t item);
 
+#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
+  // Returns a new location string or an empty string if all locations under current host have been accessedd
+  std::string getNewLocation(PerformData* performData, std::vector<std::string>& locations) const;
+
+  // Reschedules the transfer to be performed with a different host.
+  void tryNewHost(PerformData* performData, CURL* easy_handle);
+
+  // Retrieves content from either alien, cvmfs or local storage using a callback to CCDBApi.
+  void getLocalContent(PerformData* performData, std::string& newLocation, bool& contentRetrieved, std::vector<std::string>& locations);
+
+  // Continues a transfer via a http redirect.
+  void httpRedirect(PerformData* performData, std::string& newLocation, CURL* easy_handle);
+
+  // Continues a transfer via a redirect. The redirect can point to a local file, alien file or a http address.
+  void followRedirect(PerformData* performData, CURL* easy_handle, std::vector<std::string>& locations, bool& rescheduled, bool& contentRetrieved);
+#endif
+
   /**
    *  Is used to react to polling file descriptors in poll_handle.
    *
@@ -322,10 +388,12 @@ class CCDBDownloader
    */
   void checkMultiInfo();
 
+#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
   /**
    * Set openSocketCallback and closeSocketCallback with appropriate arguments. Stores data inside the CURL handle.
    */
   void setHandleOptions(CURL* handle, PerformData* data);
+#endif
 
   /**
    * Create structure holding information about a socket including a poll handle assigned to it

CCDB/include/CCDB/CcdbApi.h

@@ -24,18 +24,20 @@
 #include <TObject.h>
 #include <TMessage.h>
 #include "CCDB/CcdbObjectInfo.h"
-#include "CCDB/CCDBDownloader.h"
 #include <CommonUtils/ConfigurableParam.h>
 #include <type_traits>
 #include <vector>
 
 #if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
 #include "MemoryResources/MemoryResources.h"
+#include <boost/interprocess/sync/named_semaphore.hpp>
 #include <TJAlienCredentials.h>
 #else
 class TJAlienCredentials;
 #endif
 
+#include "CCDB/CCDBDownloader.h"
+
 class TFile;
 class TGrid;
 
@@ -342,14 +344,43 @@ class CcdbApi //: public DatabaseInterface
   TObject* retrieveFromTFile(std::string const& path, std::map<std::string, std::string> const& metadata, long timestamp,
                              std::map<std::string, std::string>* headers, std::string const& etag,
                              const std::string& createdNotAfter, const std::string& createdNotBefore) const;
-
 #if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
+  typedef struct RequestContext {
+    o2::pmr::vector<char>& dest;
+    std::string path;
+    std::map<std::string, std::string> const& metadata;
+    long timestamp;
+    std::map<std::string, std::string>& headers;
+    std::string etag;
+    std::string createdNotAfter;
+    std::string createdNotBefore;
+    bool considerSnapshot;
+
+    RequestContext(o2::pmr::vector<char>& d,
+                   std::map<std::string, std::string> const& m,
+                   std::map<std::string, std::string>& h)
+      : dest(d), metadata(m), headers(h) {}
+  } RequestContext;
+
+  // Stores file associated with requestContext as a snapshot.
+  void saveSnapshot(RequestContext& requestContext) const;
+
+  // Schedules download via CCDBDownloader, but doesn't perform it until mUVLoop is ran.
+  void scheduleDownload(RequestContext& requestContext, size_t* requestCounter) const;
+
+  void getFromSnapshot(bool createSnapshot, std::string const& path,
+                       long timestamp, std::map<std::string, std::string> headers,
+                       std::string& snapshotpath, o2::pmr::vector<char>& dest, int& fromSnapshot, std::string const& etag) const;
+  void releaseNamedSemaphore(boost::interprocess::named_semaphore* sem, std::string path) const;
+  boost::interprocess::named_semaphore* createNamedSempahore(std::string path) const;
   void loadFileToMemory(o2::pmr::vector<char>& dest, const std::string& path, std::map<std::string, std::string>* localHeaders = nullptr) const;
   void loadFileToMemory(o2::pmr::vector<char>& dest, std::string const& path,
                         std::map<std::string, std::string> const& metadata, long timestamp,
                         std::map<std::string, std::string>* headers, std::string const& etag,
                         const std::string& createdNotAfter, const std::string& createdNotBefore, bool considerSnapshot = true) const;
-  void navigateURLsAndLoadFileToMemory(o2::pmr::vector<char>& dest, CURL* curl_handle, std::string const& url, std::map<string, string>* headers) const;
+
+  // Loads files from alien and cvmfs into given destination.
+  bool loadLocalContentToMemory(o2::pmr::vector<char>& dest, std::string& url) const;
 
   // the failure to load the file to memory is signaled by 0 size and non-0 capacity
   static bool isMemoryFileInvalid(const o2::pmr::vector<char>& v) { return v.size() == 0 && v.capacity() > 0; }
@@ -364,12 +395,36 @@ class CcdbApi //: public DatabaseInterface
     }
     return obj;
   }
+
+  /**
+   * Retrieves files either as snapshot or schedules them to be downloaded via CCDBDownloader.
+   *
+   * @param requestContext Structure giving details about the transfer.
+   * @param fromSnapshot After navigateSourcesAndLoadFile returns signals whether file was retrieved from snapshot.
+   * @param requestCounter Pointer to the variable storing the number of requests to be done.
+   */
+  void navigateSourcesAndLoadFile(RequestContext& requestContext, int& fromSnapshot, size_t* requestCounter) const;
+
+  /**
+   * Retrieves files described via RequestContexts into memory. Downloads are performed in parallel via CCDBDownloader.
+   *
+   * @param requestContext Structure giving details about the transfer.
+   */
+  void vectoredLoadFileToMemory(std::vector<RequestContext>& requestContext) const;
 #endif
 
  private:
   // Sets the unique agent ID
   void setUniqueAgentID();
 
+  /**
+   * Schedules download of data associated with the curl_handle. Doing that increments the requestCounter by 1. Requests are performed by running the mUVLoop
+   *
+   * @param handle CURL handle associated with the request.
+   * @param requestCounter Pointer to the variable storing the number of requests to be done.
+   */
+  void asynchPerform(CURL* handle, size_t* requestCounter) const;
+
   // internal helper function to update a CCDB file with meta information
   static void updateMetaInformationInLocalFile(std::string const& filename, std::map<std::string, std::string> const* headers, CCDBQuery const* querysummary = nullptr);
 
@@ -550,7 +605,7 @@ class CcdbApi //: public DatabaseInterface
   CURLcode CURL_perform(CURL* handle) const;
 
   mutable CCDBDownloader* mDownloader = nullptr; //! the multi-handle (async) CURL downloader
-  bool mIsCCDBDownloaderEnabled = false;
+  bool mIsCCDBDownloaderPreferred = false;
   /// Base URL of the CCDB (with port)
   std::string mUniqueAgentID{}; // Unique User-Agent ID communicated to server for logging
   std::string mUrl{};