fetch.h

Go to the documentation of this file.

#ifndef CVMFS_FETCH_H_
#define CVMFS_FETCH_H_

#include <pthread.h>

#include <map>
#include <string>
#include <vector>

#include "cache.h"
#include "crypto/hash.h"
#include "gtest/gtest_prod.h"
#include "network/download.h"
#include "network/sink.h"

class BackoffThrottle;

namespace perf {
class Statistics;
}

namespace cvmfs {

class TransactionSink : public Sink {
 public:
  TransactionSink(CacheManager *cache_mgr, void *open_txn)
    : cache_mgr_(cache_mgr)
    , open_txn_(open_txn)
  { }
  virtual ~TransactionSink() { }
  virtual int64_t Write(const void *buf, uint64_t sz) {
    return cache_mgr_->Write(buf, sz, open_txn_);
  }
  virtual int Reset() {
    return cache_mgr_->Reset(open_txn_);
  }

 private:
  CacheManager *cache_mgr_;
  void *open_txn_;
};


class Fetcher : SingleCopy {
  FRIEND_TEST(T_Fetcher, GetTls);
  FRIEND_TEST(T_Fetcher, SignalWaitingThreads);
  friend void *TestGetTls(void *data);
  friend void *TestFetchCollapse(void *data);
  friend void *TestFetchCollapse2(void *data);
  friend void TLSDestructor(void *data);

 public:
  Fetcher(CacheManager *cache_mgr,
          download::DownloadManager *download_mgr,
          BackoffThrottle *backoff_throttle,
          perf::StatisticsTemplate statistics,
          bool external_data = false);
  ~Fetcher();
  // TODO(jblomer): reduce number of arguments
  int Fetch(const shash::Any &id,
            const uint64_t size,
            const std::string &name,
            const zlib::Algorithms compression_algorithm,
            const CacheManager::ObjectType object_type,
            const std::string &alt_url = "",
            off_t range_offset = -1);

  CacheManager *cache_mgr() { return cache_mgr_; }
  download::DownloadManager *download_mgr() { return download_mgr_; }

 private:
  struct ThreadLocalStorage {
    ThreadLocalStorage() {
      pipe_wait[0] = -1;
      pipe_wait[1] = -1;
      fetcher = NULL;
    }

    Fetcher *fetcher;
    int pipe_wait[2];
    std::vector<int> other_pipes_waiting;
    download::JobInfo download_job;
  };

  typedef std::map< shash::Any, std::vector<int> * > ThreadQueues;

  ThreadLocalStorage *GetTls();
  void CleanupTls(ThreadLocalStorage *tls);
  void SignalWaitingThreads(const int fd, const shash::Any &id,
                            ThreadLocalStorage *tls);
  int OpenSelect(const shash::Any &id,
                 const std::string &name,
                 const CacheManager::ObjectType object_type);

  bool external_;

  pthread_key_t thread_local_storage_;

  ThreadQueues queues_download_;
  pthread_mutex_t *lock_queues_download_;

  std::vector<ThreadLocalStorage *> tls_blocks_;
  pthread_mutex_t *lock_tls_blocks_;

  CacheManager *cache_mgr_;
  download::DownloadManager *download_mgr_;
  BackoffThrottle *backoff_throttle_;
  perf::Counter *n_downloads;
  perf::Counter *n_invocations;
};

}  // namespace cvmfs

#endif  // CVMFS_FETCH_H_