CCDBDownloader.h

Go to the documentation of this file.

// Copyright 2019-2023 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// In applying this license CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
#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>
#include <string>
#include <vector>
#include <iostream>
#include <thread>
#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;
typedef struct uv_poll_s uv_poll_t;
typedef struct uv_signal_s uv_signal_t;
typedef struct uv_async_s uv_async_t;
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::vector<std::string> locations;
  std::string path;
  long timestamp;
  HeaderObjectPair_t hoPair;
  std::map<std::string, std::string>* headers;
  std::string userAgent;
  // One header list per entry of `hosts`, parallel to it. Per host and not one
  // shared list because the gate token a broker expects is per endpoint: a
  // multi-host pool can mix them, and tryNewHost() swapping only the URL left
  // the second host receiving the first host's token -- answered 401, so the
  // failover silently retrieved nothing (testCcdbApi multi_host_test).
  std::vector<curl_slist*> optionsLists;
 
  std::function<bool(std::string)> localContentCallback;
} 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.
*/
 
void closeHandles(uv_handle_t* handle, void* arg);
 
curl_socket_t opensocketCallback(void* clientp, curlsocktype purpose, struct curl_sockaddr* address);
 
void onUVClose(uv_handle_t* handle);
 
enum DownloaderErrorLevel {
  MINOR,
  SEVERE
};
 
class CCDBDownloader
{
 public:
  std::unordered_map<curl_socket_t, uv_timer_t*> mSocketTimerMap;
 
  uv_loop_t* mUVLoop;
 
  std::unordered_map<uv_handle_t*, bool> mHandleMap;
 
  int mKeepaliveTimeoutMS = 100;
 
  int mConnectionTimeoutMS = 60000;
 
  int mRequestTimeoutMS = 300000;
 
  int mHappyEyeballsHeadstartMS = 500;
 
  int mMaxHandlesInUse = 3;
 
  bool mExternalLoop;
 
  CCDBDownloader(uv_loop_t* uv_loop = nullptr);
  ~CCDBDownloader();
 
  CURLcode perform(CURL* handle);
 
  std::vector<CURLcode> batchBlockingPerform(std::vector<CURL*> const& handleVector);
 
  void asynchSchedule(CURL* handle, size_t* requestCounter);
 
  void setMaxParallelConnections(int limit);
 
  void setKeepaliveTimeoutTime(int timeoutMS);
 
  void setConnectionTimeoutTime(int timeoutMS);
 
  void setRequestTimeoutTime(int timeoutMS);
 
  void setHappyEyeballsHeadstartTime(int headstartMS);
 
  void setOfflineTimeoutSettings();
 
  void setOnlineTimeoutSettings();
 
  void runLoop(bool noWait);
 
  std::string prepareLogMessage(std::string host_url, std::string userAgent, const std::string& path, long ts, const std::map<std::string, std::string>* headers, long httpCode) const;
 
  std::string trimHostUrl(std::string full_host_url) const;
 
 private:
  std::string prepareRedirectedURL(std::string address, std::string potentialHost) const;
 
  void updateLocations(std::multimap<std::string, std::string>* headerMap, std::vector<std::string>* locations, int* locIndex) const;
 
  std::string mUserAgentId = "CCDBDownloader";
  void setupInternalUVLoop();
 
  int mHandlesInUse = 0;
 
  CURLM* mCurlMultiHandle = nullptr;
 
  uv_timer_t* mTimeoutTimer;
 
  std::vector<CURL*> mHandlesToBeAdded;
 
  enum RequestType {
    BLOCKING,
    ASYNCHRONOUS
  };
 
  typedef struct curl_context_s {
    uv_poll_t* poll_handle;
    curl_socket_t sockfd = -1;
    CCDBDownloader* CD = nullptr;
  } curl_context_t;
 
  typedef struct DataForSocket {
    CCDBDownloader* CD;
    CURLM* curlm;
  } DataForSocket;
 
  DataForSocket mSocketData;
 
#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
  typedef struct PerformData {
    CURLcode* codeDestination;
    size_t* requestsLeft;
    RequestType type;
    int hostInd;
    int locInd;
    DownloaderRequestData* requestData;
    // Freed by transferFinished; indexed by hostInd, see DownloaderRequestData.
    std::vector<curl_slist*>* options;
  } PerformData;
#endif
 
  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
 
  static void curlPerform(uv_poll_t* handle, int status, int events);
 
  static int handleSocket(CURL* easy, curl_socket_t s, int action, void* userp, void* socketp);
 
  static void closeSocketByTimer(uv_timer_t* handle);
 
  static void curlTimeout(uv_timer_t* req);
 
  static void curlCloseCB(uv_handle_t* handle);
 
  static void destroyCurlContext(curl_context_t* context);
 
  static int startTimeout(CURLM* multi, long timeout_ms, void* userp);
 
  void initializeMultiHandle();
 
  void transferFinished(CURL* handle, CURLcode curlCode);
 
  void checkMultiInfo();
 
#if !defined(__CINT__) && !defined(__MAKECINT__) && !defined(__ROOTCLING__) && !defined(__CLING__)
  void setHandleOptions(CURL* handle, PerformData* data);
#endif
 
  curl_context_t* createCurlContext(curl_socket_t sockfd);
 
  void checkHandleQueue();
};
 
typedef struct DataForClosingSocket {
  CCDBDownloader* CD;
  curl_socket_t socket;
} DataForClosingSocket;
 
} // namespace o2::ccdb
 
#endif // O2_CCDB_CCDBDOWNLOADER_H