download.h

Go to the documentation of this file.

#ifndef CVMFS_DOWNLOAD_H_
#define CVMFS_DOWNLOAD_H_

#include <poll.h>
#include <pthread.h>
#include <stdint.h>
#include <unistd.h>

#include <cstdio>
#include <set>
#include <string>
#include <vector>

#include "gtest/gtest_prod.h"

#include "atomic.h"
#include "compression.h"
#include "dns.h"
#include "duplex_curl.h"
#include "hash.h"
#include "prng.h"
#include "sink.h"
#include "statistics.h"


namespace download {

enum Failures {
  kFailOk = 0,
  kFailLocalIO,
  kFailBadUrl,
  kFailProxyResolve,
  kFailHostResolve,
  // artificial failure code.  Try other host even though
  // failure seems to be at the proxy
  kFailHostAfterProxy,
  kFailProxyConnection,
  kFailHostConnection,
  kFailProxyHttp,
  kFailHostHttp,
  kFailBadData,
  kFailTooBig,
  kFailOther,
  kFailUnsupportedProtocol,
  kFailProxyTooSlow,
  kFailHostTooSlow,
  kFailProxyShortTransfer,
  kFailHostShortTransfer,

  kFailNumEntries
};  // Failures


inline bool IsHostTransferError(const Failures error) {
  switch (error) {
    case kFailHostConnection:
    case kFailHostTooSlow:
    case kFailHostShortTransfer:
      return true;
    default:
      break;
  }
  return false;
}

inline bool IsProxyTransferError(const Failures error) {
  switch (error) {
    case kFailProxyConnection:
    case kFailProxyTooSlow:
    case kFailProxyShortTransfer:
      return true;
    default:
      break;
  }
  return false;
}

inline const char *Code2Ascii(const Failures error) {
  const char *texts[kFailNumEntries + 1];
  texts[0] = "OK";
  texts[1] = "local I/O failure";
  texts[2] = "malformed URL";
  texts[3] = "failed to resolve proxy address";
  texts[4] = "failed to resolve host address";
  texts[5] = "all proxies failed, trying host fail-over";
  texts[6] = "proxy connection problem";
  texts[7] = "host connection problem";
  texts[8] = "proxy returned HTTP error";
  texts[9] = "host returned HTTP error";
  texts[10] = "corrupted data received";
  texts[11] = "resource too big to download";
  texts[12] = "unknown network error";
  texts[13] = "Unsupported URL in protocol";
  texts[14] = "proxy serving data too slowly";
  texts[15] = "host serving data too slowly";
  texts[16] = "proxy data transfer cut short";
  texts[17] = "host data transfer cut short";
  texts[18] = "no text";
  return texts[error];
}

enum Destination {
  kDestinationMem = 1,
  kDestinationFile,
  kDestinationPath,
  kDestinationSink,
  kDestinationNone
};  // Destination


struct Counters {
  perf::Counter *sz_transferred_bytes;
  perf::Counter *sz_transfer_time;  // measured in miliseconds
  perf::Counter *n_requests;
  perf::Counter *n_retries;
  perf::Counter *n_proxy_failover;
  perf::Counter *n_host_failover;

  explicit Counters(perf::StatisticsTemplate statistics) {
    sz_transferred_bytes = statistics.RegisterTemplated("sz_transferred_bytes",
        "Number of transferred bytes");
    sz_transfer_time = statistics.RegisterTemplated("sz_transfer_time",
        "Transfer time (miliseconds)");
    n_requests = statistics.RegisterTemplated("n_requests",
        "Number of requests");
    n_retries = statistics.RegisterTemplated("n_retries", "Number of retries");
    n_proxy_failover = statistics.RegisterTemplated("n_proxy_failover",
        "Number of proxy failovers");
    n_host_failover = statistics.RegisterTemplated("n_host_failover",
        "Number of host failovers");
  }
};  // Counters


struct JobInfo {
  const std::string *url;
  bool compressed;
  bool probe_hosts;
  bool head_request;
  bool follow_redirects;
  bool force_nocache;
  pid_t pid;
  uid_t uid;
  gid_t gid;
  void *cred_data;  // Per-transfer credential data
  Destination destination;
  struct {
    size_t size;
    size_t pos;
    char *data;
  } destination_mem;
  FILE *destination_file;
  const std::string *destination_path;
  cvmfs::Sink *destination_sink;
  const shash::Any *expected_hash;
  const std::string *extra_info;

  // Allow byte ranges to be specified.
  off_t range_offset;
  off_t range_size;

  // Default initialization of fields
  void Init() {
    url = NULL;
    compressed = false;
    probe_hosts = false;
    head_request = false;
    follow_redirects = false;
    force_nocache = false;
    pid = -1;
    uid = -1;
    gid = -1;
    cred_data = NULL;
    destination = kDestinationNone;
    destination_mem.size = destination_mem.pos = 0;
    destination_mem.data = NULL;
    destination_file = NULL;
    destination_path = NULL;
    destination_sink = NULL;
    expected_hash = NULL;
    extra_info = NULL;

    curl_handle = NULL;
    headers = NULL;
    memset(&zstream, 0, sizeof(zstream));
    info_header = NULL;
    wait_at[0] = wait_at[1] = -1;
    nocache = false;
    error_code = kFailOther;
    num_used_proxies = num_used_hosts = num_retries = 0;
    backoff_ms = 0;
    current_host_chain_index = 0;

    range_offset = -1;
    range_size = -1;
    http_code = -1;
  }

  // One constructor per destination + head request
  JobInfo() { Init(); }
  JobInfo(const std::string *u, const bool c, const bool ph,
          const std::string *p, const shash::Any *h)
  {
    Init();
    url = u;
    compressed = c;
    probe_hosts = ph;
    destination = kDestinationPath;
    destination_path = p;
    expected_hash = h;
  }
  JobInfo(const std::string *u, const bool c, const bool ph, FILE *f,
          const shash::Any *h)
  {
    Init();
    url = u;
    compressed = c;
    probe_hosts = ph;
    destination = kDestinationFile;
    destination_file = f;
    expected_hash = h;
  }
  JobInfo(const std::string *u, const bool c, const bool ph,
          const shash::Any *h)
  {
    Init();
    url = u;
    compressed = c;
    probe_hosts = ph;
    destination = kDestinationMem;
    expected_hash = h;
  }
  JobInfo(const std::string *u, const bool c, const bool ph,
          cvmfs::Sink *s, const shash::Any *h)
  {
    Init();
    url = u;
    compressed = c;
    probe_hosts = ph;
    destination = kDestinationSink;
    destination_sink = s;
    expected_hash = h;
  }
  JobInfo(const std::string *u, const bool ph) {
    Init();
    url = u;
    probe_hosts = ph;
    head_request = true;
  }

  ~JobInfo() {
    if (wait_at[0] >= 0) {
      close(wait_at[0]);
      close(wait_at[1]);
    }
  }

  bool IsFileNotFound();

  // Internal state, don't touch
  CURL *curl_handle;
  curl_slist *headers;
  char *info_header;
  z_stream zstream;
  shash::ContextPtr hash_context;
  int wait_at[2];  
  std::string proxy;
  bool nocache;
  Failures error_code;
  int http_code;
  unsigned char num_used_proxies;
  unsigned char num_used_hosts;
  unsigned char num_retries;
  unsigned backoff_ms;
  unsigned int current_host_chain_index;
};  // JobInfo


class HeaderLists {
  FRIEND_TEST(T_HeaderLists, Intrinsics);
 public:
  ~HeaderLists();
  curl_slist *GetList(const char *header);
  curl_slist *DuplicateList(curl_slist *slist);
  void AppendHeader(curl_slist *slist, const char *header);
  void CutHeader(const char *header, curl_slist **slist);
  void PutList(curl_slist *slist);
  std::string Print(curl_slist *slist);

 private:
  static const unsigned kBlockSize = 4096/sizeof(curl_slist);

  bool IsUsed(curl_slist *slist) { return slist->data != NULL; }
  curl_slist *Get(const char *header);
  void Put(curl_slist *slist);
  void AddBlock();

  std::vector<curl_slist *> blocks_;  // List of curl_slist blocks
};


class CredentialsAttachment {
 public:
  virtual ~CredentialsAttachment() { }
  virtual bool ConfigureCurlHandle(CURL *curl_handle,
                                   pid_t pid,
                                   void **info_data) = 0;
  virtual void ReleaseCurlHandle(CURL *curl_handle, void *info_data) = 0;
};


class DownloadManager {
  FRIEND_TEST(T_Download, ValidateGeoReply);
  FRIEND_TEST(T_Download, StripDirect);

 public:
  struct ProxyInfo {
    ProxyInfo() { }
    explicit ProxyInfo(const std::string &url) : url(url) { }
    ProxyInfo(const dns::Host &host, const std::string &url)
      : host(host)
      , url(url)
    { }
    std::string Print();
    dns::Host host;
    std::string url;
  };

  enum ProxySetModes {
    kSetProxyRegular = 0,
    kSetProxyFallback,
    kSetProxyBoth,
  };

  static const int kProbeUnprobed;
  static const int kProbeDown;
  static const int kProbeGeo;

  static const unsigned kMaxMemSize;

  static const unsigned kDnsDefaultRetries = 1;
  static const unsigned kDnsDefaultTimeoutMs = 3000;

  DownloadManager();
  ~DownloadManager();

  static int ParseHttpCode(const char digits[3]);

  void Init(const unsigned max_pool_handles,
            const bool use_system_proxy,
            perf::StatisticsTemplate statistics);
  void Fini();
  void Spawn();
  DownloadManager *Clone(perf::StatisticsTemplate statistics);
  Failures Fetch(JobInfo *info);

  void SetCredentialsAttachment(CredentialsAttachment *ca);
  std::string GetDnsServer() const;
  void SetDnsServer(const std::string &address);
  void SetDnsParameters(const unsigned retries, const unsigned timeout_ms);
  void SetDnsTtlLimits(const unsigned min_seconds, const unsigned max_seconds);
  void SetIpPreference(const dns::IpPreference preference);
  void SetTimeout(const unsigned seconds_proxy, const unsigned seconds_direct);
  void GetTimeout(unsigned *seconds_proxy, unsigned *seconds_direct);
  void SetLowSpeedLimit(const unsigned low_speed_limit);
  void SetHostChain(const std::string &host_list);
  void SetHostChain(const std::vector<std::string> &host_list);
  void GetHostInfo(std::vector<std::string> *host_chain,
                   std::vector<int> *rtt, unsigned *current_host);
  void ProbeHosts();
  bool ProbeGeo();
    // Sort list of servers using the Geo API.  If the output_order
    // vector is NULL, then the servers vector input is itself sorted.
    // If it is non-NULL, then servers is left unchanged and the zero-based
    // ordering is stored into output_order.
  bool GeoSortServers(std::vector<std::string> *servers,
                      std::vector<uint64_t>    *output_order = NULL);
  void SwitchHost();
  void SetProxyChain(const std::string &proxy_list,
                     const std::string &fallback_proxy_list,
                     const ProxySetModes set_mode);
  void GetProxyInfo(std::vector< std::vector<ProxyInfo> > *proxy_chain,
                    unsigned *current_group,
                    unsigned *fallback_group);
  std::string GetProxyList();
  std::string GetFallbackProxyList();
  void RebalanceProxies();
  void SwitchProxyGroup();
  void SetProxyGroupResetDelay(const unsigned seconds);
  void SetHostResetDelay(const unsigned seconds);
  void SetRetryParameters(const unsigned max_retries,
                          const unsigned backoff_init_ms,
                          const unsigned backoff_max_ms);
  void SetMaxIpaddrPerProxy(unsigned limit);
  void SetProxyTemplates(const std::string &direct, const std::string &forced);
  void EnableInfoHeader();
  void EnableRedirects();

  unsigned num_hosts() {
    if (opt_host_chain_) return opt_host_chain_->size();
    return 0;
  }

  dns::IpPreference opt_ip_preference() const {
    return opt_ip_preference_;
  }

 private:
  static int CallbackCurlSocket(CURL *easy, curl_socket_t s, int action,
                                void *userp, void *socketp);
  static void *MainDownload(void *data);

  bool StripDirect(const std::string &proxy_list, std::string *cleaned_list);
  bool ValidateGeoReply(const std::string &reply_order,
                        const unsigned expected_size,
                        std::vector<uint64_t> *reply_vals);
  void SwitchHost(JobInfo *info);
  void SwitchProxy(JobInfo *info);
  void RebalanceProxiesUnlocked();
  CURL *AcquireCurlHandle();
  void ReleaseCurlHandle(CURL *handle);
  void ReleaseCredential(JobInfo *info);
  void InitializeRequest(JobInfo *info, CURL *handle);
  void SetUrlOptions(JobInfo *info);
  void ValidateProxyIpsUnlocked(const std::string &url, const dns::Host &host);
  void UpdateStatistics(CURL *handle);
  bool CanRetry(const JobInfo *info);
  void Backoff(JobInfo *info);
  void SetNocache(JobInfo *info);
  void SetRegularCache(JobInfo *info);
  bool VerifyAndFinalize(const int curl_error, JobInfo *info);
  void InitHeaders();
  void FiniHeaders();
  void CloneProxyConfig(DownloadManager *clone);

  Prng prng_;
  std::set<CURL *> *pool_handles_idle_;
  std::set<CURL *> *pool_handles_inuse_;
  uint32_t pool_max_handles_;
  CURLM *curl_multi_;
  HeaderLists *header_lists_;
  curl_slist *default_headers_;
  char *user_agent_;

  pthread_t thread_download_;
  atomic_int32 multi_threaded_;
  int pipe_terminate_[2];

  int pipe_jobs_[2];
  struct pollfd *watch_fds_;
  uint32_t watch_fds_size_;
  uint32_t watch_fds_inuse_;
  uint32_t watch_fds_max_;

  pthread_mutex_t *lock_options_;
  pthread_mutex_t *lock_synchronous_mode_;
  std::string opt_dns_server_;
  unsigned opt_timeout_proxy_;
  unsigned opt_timeout_direct_;
  unsigned opt_low_speed_limit_;
  unsigned opt_max_retries_;
  unsigned opt_backoff_init_ms_;
  unsigned opt_backoff_max_ms_;
  bool enable_info_header_;
  bool opt_ipv4_only_;
  bool follow_redirects_;
  bool use_system_proxy_;

  // Host list
  std::vector<std::string> *opt_host_chain_;
  std::vector<int> *opt_host_chain_rtt_;
  unsigned opt_host_chain_current_;

  // Proxy list
  std::vector< std::vector<ProxyInfo> > *opt_proxy_groups_;
  unsigned opt_proxy_groups_current_;
  unsigned opt_proxy_groups_current_burned_;
  unsigned opt_proxy_groups_fallback_;
  unsigned opt_num_proxies_;
  std::string opt_proxy_list_;
  std::string opt_proxy_fallback_list_;

  dns::NormalResolver *resolver_;

  dns::IpPreference opt_ip_preference_;

  std::string proxy_template_direct_;
  std::string proxy_template_forced_;

  time_t opt_timestamp_backup_proxies_;
  time_t opt_timestamp_failover_proxies_;  // failover within the same group
  unsigned opt_proxy_groups_reset_after_;

  time_t opt_timestamp_backup_host_;
  unsigned opt_host_reset_after_;

  CredentialsAttachment *credentials_attachment_;

  // Writes and reads should be atomic because reading happens in a different
  // thread than writing.
  Counters *counters_;
};  // DownloadManager

}  // namespace download

#endif  // CVMFS_DOWNLOAD_H_