Head

GCC Code Coverage Report

Directory:	cvmfs/		Exec	Total	Coverage
File:	cvmfs/fetch.h	Lines:	16	16	100.0 %
Date:	2019-02-03 02:48:13	Branches:	1	2	50.0 %


/**
 * This file is part of the CernVM File System.
 */

#ifndef CVMFS_FETCH_H_
#define CVMFS_FETCH_H_

#include <pthread.h>

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

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

class BackoffThrottle;

namespace perf {
class Statistics;
}

namespace cvmfs {

/**
 * TransacionSink uses an open transaction in a cache manager as a sink.  It
 * allows the download manager to write data without knowing about the cache
 * manager.
 */
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_;
};


/**
 * The Fetcher uses a cache manager and a download manager in order to provide a
 * (virtual) file descriptor to a requested object, which is valid in the
 * context of the cache manager.
 * If the object is not in the cache, it is downloaded and stored in the cache.
 *
 * Concurrent download requests for the same id are collapsed.
 */
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:
  /**
   * Multiple threads might want to download the same object at the same time.
   * If that happens, only the first thread performs the download.  The other
   * threads wait on a pipe for a notification from the first thread.
   */
  struct ThreadLocalStorage {
    ThreadLocalStorage() {
      pipe_wait[0] = -1;
      pipe_wait[1] = -1;
      fetcher = NULL;
    }

    /**
     * Used during cleanup to find tls_blocks_.
     */
    Fetcher *fetcher;
    /**
     * Wait on the reading end if another thread is already downloading the same
     * object.
     */
    int pipe_wait[2];
    /**
     * Writer ends of all the pipes of threads that want to download the same
     * object.
     */
    std::vector<int> other_pipes_waiting;
    /**
     * It is sufficient to construct the JobInfo object once per thread, not
     * on every call to Fetch().
     */
    download::JobInfo download_job;
  };

  /**
   * Maps currently downloaded chunks to the other_pipes_waiting member of the
   * thread local storage of the downloading thread.  This way, a thread can
   * enqueue itself to such an other_pipes_waiting list and gets informed when
   * the download is completed.
   */
  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);

  /**
   * If set to true, this fetcher is in 'external data' mode:
   * instead of constructing the to-be-downloaded URL from the entry hash,
   * it will use the filename.
   */
  bool external_;

  /**
   * Key to the thread's ThreadLocalStorage memory
   */
  pthread_key_t thread_local_storage_;

  ThreadQueues queues_download_;
  pthread_mutex_t *lock_queues_download_;

  /**
   * All the threads register their thread local storage here, so that it can
   * be cleaned up properly in the destructor of Fetcher.
   */
  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;
};

}  // namespace cvmfs

#endif  // CVMFS_FETCH_H_


Generated by: GCOVR (Version 4.1)

Line	Branch	Exec	Source
1			/**
2			* This file is part of the CernVM File System.
3			*/
4
5			#ifndef CVMFS_FETCH_H_
6			#define CVMFS_FETCH_H_
7
8			#include <pthread.h>
9
10			#include <map>
11			#include <string>
12			#include <vector>
13
14			#include "cache.h"
15			#include "download.h"
16			#include "gtest/gtest_prod.h"
17			#include "hash.h"
18			#include "sink.h"
19
20			class BackoffThrottle;
21
22			namespace perf {
23			class Statistics;
24			}
25
26			namespace cvmfs {
27
28			/**
29			* TransacionSink uses an open transaction in a cache manager as a sink. It
30			* allows the download manager to write data without knowing about the cache
31			* manager.
32			*/
33			class TransactionSink : public Sink {
34			public:
35		98	TransactionSink(CacheManager cache_mgr, void open_txn)
36			: cache_mgr_(cache_mgr)
37		98	, open_txn_(open_txn)
38		98	{ }
39	✗✓	98	virtual ~TransactionSink() { }
40		100	virtual int64_t Write(const void *buf, uint64_t sz) {
41		100	return cache_mgr_->Write(buf, sz, open_txn_);
42			}
43		2	virtual int Reset() {
44		2	return cache_mgr_->Reset(open_txn_);
45			}
46
47			private:
48			CacheManager *cache_mgr_;
49			void *open_txn_;
50			};
51
52
53			/**
54			* The Fetcher uses a cache manager and a download manager in order to provide a
55			* (virtual) file descriptor to a requested object, which is valid in the
56			* context of the cache manager.
57			* If the object is not in the cache, it is downloaded and stored in the cache.
58			*
59			* Concurrent download requests for the same id are collapsed.
60			*/
61			class Fetcher : SingleCopy {
62			FRIEND_TEST(T_Fetcher, GetTls);
63			FRIEND_TEST(T_Fetcher, SignalWaitingThreads);
64			friend void TestGetTls(void data);
65			friend void TestFetchCollapse(void data);
66			friend void TestFetchCollapse2(void data);
67			friend void TLSDestructor(void *data);
68
69			public:
70			Fetcher(CacheManager *cache_mgr,
71			download::DownloadManager *download_mgr,
72			BackoffThrottle *backoff_throttle,
73			perf::StatisticsTemplate statistics,
74			bool external_data = false);
75			~Fetcher();
76			// TODO(jblomer): reduce number of arguments
77			int Fetch(const shash::Any &id,
78			const uint64_t size,
79			const std::string &name,
80			const zlib::Algorithms compression_algorithm,
81			const CacheManager::ObjectType object_type,
82			const std::string &alt_url = "",
83			off_t range_offset = -1);
84
85		103	CacheManager *cache_mgr() { return cache_mgr_; }
86		24	download::DownloadManager *download_mgr() { return download_mgr_; }
87
88			private:
89			/**
90			* Multiple threads might want to download the same object at the same time.
91			* If that happens, only the first thread performs the download. The other
92			* threads wait on a pipe for a notification from the first thread.
93			*/
94		86	struct ThreadLocalStorage {
95		86	ThreadLocalStorage() {
96		86	pipe_wait[0] = -1;
97		86	pipe_wait[1] = -1;
98		86	fetcher = NULL;
99		86	}
100
101			/**
102			* Used during cleanup to find tls_blocks_.
103			*/
104			Fetcher *fetcher;
105			/**
106			* Wait on the reading end if another thread is already downloading the same
107			* object.
108			*/
109			int pipe_wait[2];
110			/**
111			* Writer ends of all the pipes of threads that want to download the same
112			* object.
113			*/
114			std::vector<int> other_pipes_waiting;
115			/**
116			* It is sufficient to construct the JobInfo object once per thread, not
117			* on every call to Fetch().
118			*/
119			download::JobInfo download_job;
120			};
121
122			/**
123			* Maps currently downloaded chunks to the other_pipes_waiting member of the
124			* thread local storage of the downloading thread. This way, a thread can
125			* enqueue itself to such an other_pipes_waiting list and gets informed when
126			* the download is completed.
127			*/
128			typedef std::map< shash::Any, std::vector<int> * > ThreadQueues;
129
130			ThreadLocalStorage *GetTls();
131			void CleanupTls(ThreadLocalStorage *tls);
132			void SignalWaitingThreads(const int fd, const shash::Any &id,
133			ThreadLocalStorage *tls);
134			int OpenSelect(const shash::Any &id,
135			const std::string &name,
136			const CacheManager::ObjectType object_type);
137
138			/**
139			* If set to true, this fetcher is in 'external data' mode:
140			* instead of constructing the to-be-downloaded URL from the entry hash,
141			* it will use the filename.
142			*/
143			bool external_;
144
145			/**
146			* Key to the thread's ThreadLocalStorage memory
147			*/
148			pthread_key_t thread_local_storage_;
149
150			ThreadQueues queues_download_;
151			pthread_mutex_t *lock_queues_download_;
152
153			/**
154			* All the threads register their thread local storage here, so that it can
155			* be cleaned up properly in the destructor of Fetcher.
156			*/
157			std::vector<ThreadLocalStorage *> tls_blocks_;
158			pthread_mutex_t *lock_tls_blocks_;
159
160			CacheManager *cache_mgr_;
161			download::DownloadManager *download_mgr_;
162			BackoffThrottle *backoff_throttle_;
163			perf::Counter *n_downloads;
164			};
165
166			} // namespace cvmfs
167
168			#endif // CVMFS_FETCH_H_