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

Author: davidrohr

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;
@@ -29,11 +35,28 @@ typedef struct uv_signal_s uv_signal_t;
 typedef struct uv_async_s uv_async_t;
 typedef struct uv_handle_s uv_handle_t;
 
-using namespace std;
-
 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.
@@ -135,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.
    */
@@ -178,6 +209,14 @@ class CCDBDownloader
   void runLoop(bool noWait);
 
  private:
+  /**
+   * 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::string baseUrl, std::multimap<std::string, std::string>* headerMap) const;
+
   std::string mUserAgentId = "CCDBDownloader";
   /**
    * Sets up internal UV loop.
@@ -209,8 +248,7 @@ class CCDBDownloader
    */
   enum RequestType {
     BLOCKING,
-    ASYNCHRONOUS,
-    ASYNCHRONOUS_WITH_CALLBACK
+    ASYNCHRONOUS
   };
 
   /**
@@ -232,20 +270,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.
@@ -255,6 +292,20 @@ class CCDBDownloader
    */
   static void closesocketCallback(void* clientp, curl_socket_t item);
 
+#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
+  // 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& newUrl, std::string& newLocation, bool& contentRetrieved, std::vector<std::string>& locations);
+
+  // Continues a transfer via a http redirect.
+  void httpRedirect(PerformData* performData, std::string& newUrl, 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.
    *
@@ -324,10 +375,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;
 
@@ -119,34 +121,34 @@ class CcdbApi //: public DatabaseInterface
   static std::unique_ptr<std::vector<char>> createObjectImage(const void* obj, std::type_info const& tinfo, CcdbObjectInfo* info = nullptr);
 
   /**
-     * Store into the CCDB a TFile containing the ROOT object.
-     *
-     * @param rootObject Raw pointer to the object to store.
-     * @param path The path where the object is going to be stored.
-     * @param metadata Key-values representing the metadata for this object.
-     * @param startValidityTimestamp Start of validity. If omitted, current timestamp is used.
-     * @param endValidityTimestamp End of validity. If omitted, current timestamp + 1 day is used.
-     * @return 0 -> ok,
-     *         positive number -> curl error (https://curl.se/libcurl/c/libcurl-errors.html),
-     *         -1 : object bigger than maxSize,
-     *         -2 : curl initialization error
-     */
+   * Store into the CCDB a TFile containing the ROOT object.
+   *
+   * @param rootObject Raw pointer to the object to store.
+   * @param path The path where the object is going to be stored.
+   * @param metadata Key-values representing the metadata for this object.
+   * @param startValidityTimestamp Start of validity. If omitted, current timestamp is used.
+   * @param endValidityTimestamp End of validity. If omitted, current timestamp + 1 day is used.
+   * @return 0 -> ok,
+   *         positive number -> curl error (https://curl.se/libcurl/c/libcurl-errors.html),
+   *         -1 : object bigger than maxSize,
+   *         -2 : curl initialization error
+   */
   int storeAsTFile(const TObject* rootObject, std::string const& path, std::map<std::string, std::string> const& metadata,
                    long startValidityTimestamp = -1, long endValidityTimestamp = -1, std::vector<char>::size_type maxSize = 0 /*bytes*/) const;
 
   /**
-     * Store into the CCDB a TFile containing an object of type T (which needs to have a ROOT dictionary)
-     *
-     * @param obj Raw pointer to the object to store.
-     * @param path The path where the object is going to be stored.
-     * @param metadata Key-values representing the metadata for this object.
-     * @param startValidityTimestamp Start of validity. If omitted, current timestamp is used.
-     * @param endValidityTimestamp End of validity. If omitted, current timestamp + 1 day is used.
-     * @return 0 -> ok,
-     *         positive number -> curl error (https://curl.se/libcurl/c/libcurl-errors.html),
-     *         -1 : object bigger than maxSize,
-     *         -2 : curl initialization error
-     */
+   * Store into the CCDB a TFile containing an object of type T (which needs to have a ROOT dictionary)
+   *
+   * @param obj Raw pointer to the object to store.
+   * @param path The path where the object is going to be stored.
+   * @param metadata Key-values representing the metadata for this object.
+   * @param startValidityTimestamp Start of validity. If omitted, current timestamp is used.
+   * @param endValidityTimestamp End of validity. If omitted, current timestamp + 1 day is used.
+   * @return 0 -> ok,
+   *         positive number -> curl error (https://curl.se/libcurl/c/libcurl-errors.html),
+   *         -1 : object bigger than maxSize,
+   *         -2 : curl initialization error
+   */
   template <typename T>
   int storeAsTFileAny(const T* obj, std::string const& path, std::map<std::string, std::string> const& metadata,
                       long startValidityTimestamp = -1, long endValidityTimestamp = -1, std::vector<char>::size_type maxSize = 0 /*bytes*/) const
@@ -233,7 +235,7 @@ class CcdbApi //: public DatabaseInterface
    * @param returnFormat The format of the returned string -> one of "text/plain (default)", "application/json", "text/xml"
    * @return The listing of folder and/or objects in the format requested
    */
-  std::string list(std::string const& path = "", bool latestOnly = false, std::string const& returnFormat = "text/plain") const;
+  std::string list(std::string const& path = "", bool latestOnly = false, std::string const& returnFormat = "text/plain", long createdNotAfter = -1, long createdNotBefore = -1) const;
 
   /**
    * Make a local snapshot of all valid objects, given a timestamp, of the CCDB under a given local path.
@@ -249,12 +251,12 @@ class CcdbApi //: public DatabaseInterface
   bool isHostReachable() const;
 
   /**
-  * Helper function to extract the list of sub-folders from a list reply into a vector container.
-  * Can be used to achieve full recursive traversal/listing of the CCDB.
-  *
-  * @param reply The reply that we got from a GET/browse sort of request.
-  * @return The vector of sub-folders.
-  */
+   * Helper function to extract the list of sub-folders from a list reply into a vector container.
+   * Can be used to achieve full recursive traversal/listing of the CCDB.
+   *
+   * @param reply The reply that we got from a GET/browse sort of request.
+   * @return The vector of sub-folders.
+   */
   std::vector<std::string> parseSubFolders(std::string const& reply) const;
 
   /**
@@ -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);
 
@@ -526,9 +581,9 @@ class CcdbApi //: public DatabaseInterface
                      const std::string& createdNotAfter, const std::string& createdNotBefore, bool followRedirect, CurlWriteCallback writeCallback) const;
 
   /**
-  * Initialize hostsPool
-  * @param hosts string with hosts separated by "," or ";"
-  */
+   * Initialize hostsPool
+   * @param hosts string with hosts separated by "," or ";"
+   */
   void initHostsPool(std::string hosts);
 
   std::string getHostUrl(int hostIndex) const;
@@ -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{};
@@ -595,4 +650,4 @@ typename std::enable_if<std::is_base_of<o2::conf::ConfigurableParam, T>::value,
 } // namespace ccdb
 } // namespace o2
 
-#endif //PROJECT_CCDBAPI_H
+#endif // PROJECT_CCDBAPI_H