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)
    : Sink(false),
      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_);
  }

  virtual int Purge() {
    return Reset();
  }
  virtual bool IsValid() {
    return cache_mgr_ != NULL && open_txn_ != NULL;
  }

  virtual int Flush() { return 0; }
  virtual bool Reserve(size_t /*size*/) { return true; }
  virtual bool RequiresReserve() { return false; }

  virtual std::string Describe() {
    std::string result = "Transaction sink that is ";
    result += IsValid() ? "valid"  : "invalid";
    return result;
  }

 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);
  ~Fetcher();

  int Fetch(const CacheManager::LabeledObject &object,
            const std::string &alt_url = "");

  void ReplaceCacheManager(CacheManager *new_cache_mgr) {
    cache_mgr_ = new_cache_mgr;
  }
  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 CacheManager::LabeledObject &object);

  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_