GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/fetch.cc
Date: 2026-04-26 02:35:59
Exec Total Coverage
Lines: 155 157 98.7%
Branches: 101 173 58.4%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5
6 #include "fetch.h"
7
8 #include <unistd.h>
9
10 #include "backoff.h"
11 #include "cache.h"
12 #include "clientctx.h"
13 #include "network/download.h"
14 #include "quota.h"
15 #include "statistics.h"
16 #include "util/logging.h"
17 #include "util/posix.h"
18
19 using namespace std; // NOLINT
20
21 namespace cvmfs {
22
23 68 void TLSDestructor(void *data) {
24 68 Fetcher::ThreadLocalStorage *tls = static_cast<Fetcher::ThreadLocalStorage *>(
25 data);
26 68 std::vector<Fetcher::ThreadLocalStorage *> *tls_blocks = &tls->fetcher
27 ->tls_blocks_;
28
29 {
30 68 const MutexLockGuard m(tls->fetcher->lock_tls_blocks_);
31 68 for (vector<Fetcher::ThreadLocalStorage *>::iterator
32 68 i = tls_blocks->begin(),
33 68 iEnd = tls_blocks->end();
34
1/2
✓ Branch 1 taken 136 times.
✗ Branch 2 not taken.
 136 i != iEnd;
35 68 ++i) {
36
2/2
✓ Branch 1 taken 68 times.
✓ Branch 2 taken 68 times.
 136 if ((*i) == tls) {
37
1/2
✓ Branch 2 taken 68 times.
✗ Branch 3 not taken.
 68 tls_blocks->erase(i);
38 68 break;
39 }
40 }
41 68 }
42 68 tls->fetcher->CleanupTls(tls);
43 68 }
44
45
46 /**
47 * Called when a thread exists, releases a ThreadLocalStorage object and
48 * removes the pointer to it from tls_blocks_.
49 */
50 963 void Fetcher::CleanupTls(ThreadLocalStorage *tls) {
51 963 ClosePipe(tls->pipe_wait);
52
1/2
✓ Branch 0 taken 963 times.
✗ Branch 1 not taken.
 963 delete tls;
53 963 }
54
55
56 /**
57 * Initialized thread-local storage if called the first time in a new thread.
58 */
59 1716 Fetcher::ThreadLocalStorage *Fetcher::GetTls() {
60 ThreadLocalStorage *tls = static_cast<ThreadLocalStorage *>(
61 1716 pthread_getspecific(thread_local_storage_));
62
2/2
✓ Branch 0 taken 753 times.
✓ Branch 1 taken 963 times.
 1716 if (tls != NULL)
63 753 return tls;
64
65
2/4
✓ Branch 1 taken 963 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 963 times.
✗ Branch 5 not taken.
 963 tls = new ThreadLocalStorage();
66 963 tls->fetcher = this;
67
1/2
✓ Branch 1 taken 963 times.
✗ Branch 2 not taken.
 963 MakePipe(tls->pipe_wait);
68 963 tls->download_job.SetCompressed(true);
69 963 tls->download_job.SetProbeHosts(true);
70 963 const int retval = pthread_setspecific(thread_local_storage_, tls);
71
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 963 times.
 963 assert(retval == 0);
72
73 963 const MutexLockGuard m(lock_tls_blocks_);
74
1/2
✓ Branch 1 taken 963 times.
✗ Branch 2 not taken.
 963 tls_blocks_.push_back(tls);
75
76 963 return tls;
77 963 }
78
79
80 1962 int Fetcher::Fetch(const CacheManager::LabeledObject &object,
81 const std::string &alt_url) {
82 int fd_return; // Read-only file descriptor that is returned
83 int retval;
84
85 1962 perf::Inc(n_invocations);
86
87 // Try to open from local cache
88
3/4
✓ Branch 1 taken 1962 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 586 times.
✓ Branch 4 taken 1376 times.
 1962 if ((fd_return = OpenSelect(object)) >= 0) {
89
1/2
✓ Branch 2 taken 586 times.
✗ Branch 3 not taken.
 586 LogCvmfs(kLogCache, kLogDebug, "hit: %s", object.label.path.c_str());
90 586 return fd_return;
91 }
92
93
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 1376 times.
 1376 if (object.id.IsNull()) {
94 // This has been seen when trying to load the root catalog signed by an
95 // invalid certificate on an empty cache
96 // TODO(jblomer): check if still necessary after the catalog reload refactor
97 ✗ LogCvmfs(kLogCache, kLogDebug, "cancel attempt to download null hash");
98 ✗ return -EIO;
99 }
100
101
1/2
✓ Branch 1 taken 1376 times.
✗ Branch 2 not taken.
 1376 ThreadLocalStorage *tls = GetTls();
102
103 // Synchronization point: either act as a master thread for this object or
104 // enqueue to the list of waiting threads.
105 1376 pthread_mutex_lock(lock_queues_download_);
106 1376 const ThreadQueues::iterator iDownloadQueue = queues_download_.find(
107
1/2
✓ Branch 1 taken 1376 times.
✗ Branch 2 not taken.
 1376 object.id);
108
2/2
✓ Branch 2 taken 34 times.
✓ Branch 3 taken 1342 times.
 1376 if (iDownloadQueue != queues_download_.end()) {
109
1/2
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
 34 LogCvmfs(kLogCache, kLogDebug, "waiting for download of %s",
110 object.label.path.c_str());
111
112
1/2
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
 34 iDownloadQueue->second->push_back(tls->pipe_wait[1]);
113 34 pthread_mutex_unlock(lock_queues_download_);
114
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 ReadPipe(tls->pipe_wait[0], &fd_return, sizeof(int));
115
116
1/2
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
 34 LogCvmfs(kLogCache, kLogDebug, "received from another thread fd %d for %s",
117 fd_return, object.label.path.c_str());
118 34 return fd_return;
119 } else {
120 // Seems we are the first one, check again in the cache (race condition)
121
1/2
✓ Branch 1 taken 1342 times.
✗ Branch 2 not taken.
 1342 fd_return = OpenSelect(object);
122
2/2
✓ Branch 0 taken 34 times.
✓ Branch 1 taken 1308 times.
 1342 if (fd_return >= 0) {
123 34 pthread_mutex_unlock(lock_queues_download_);
124 34 return fd_return;
125 }
126
127 // Create a new queue for this chunk
128
1/2
✓ Branch 1 taken 1308 times.
✗ Branch 2 not taken.
 1308 queues_download_[object.id] = &tls->other_pipes_waiting;
129 1308 pthread_mutex_unlock(lock_queues_download_);
130 }
131
132 1308 perf::Inc(n_downloads);
133
134 // Involve the download manager
135
1/2
✓ Branch 2 taken 1308 times.
✗ Branch 3 not taken.
 1308 LogCvmfs(kLogCache, kLogDebug, "downloading %s", object.label.path.c_str());
136 1308 std::string url;
137
2/2
✓ Branch 1 taken 102 times.
✓ Branch 2 taken 1206 times.
 1308 if (object.label.IsExternal()) {
138
2/4
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 102 times.
✗ Branch 5 not taken.
 102 url = !alt_url.empty() ? alt_url : object.label.path;
139 } else {
140
8/14
✓ Branch 1 taken 34 times.
✓ Branch 2 taken 1172 times.
✓ Branch 4 taken 34 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 1172 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 1172 times.
✗ Branch 11 not taken.
✓ Branch 13 taken 1206 times.
✗ Branch 14 not taken.
✓ Branch 18 taken 1172 times.
✓ Branch 19 taken 34 times.
✗ Branch 22 not taken.
✗ Branch 23 not taken.
 1206 url = "/" + (alt_url.size() ? alt_url : "data/" + object.id.MakePath());
141 }
142
1/2
✓ Branch 1 taken 1308 times.
✗ Branch 2 not taken.
 1308 void *txn = alloca(cache_mgr_->SizeOfTxn());
143
1/2
✓ Branch 1 taken 1308 times.
✗ Branch 2 not taken.
 1308 retval = cache_mgr_->StartTxn(object.id, object.label.size, txn);
144
2/2
✓ Branch 0 taken 34 times.
✓ Branch 1 taken 1274 times.
 1308 if (retval < 0) {
145
1/2
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
 34 LogCvmfs(kLogCache, kLogDebug, "could not start transaction on %s",
146 object.label.path.c_str());
147
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 SignalWaitingThreads(retval, object.id, tls);
148 34 return retval;
149 }
150
1/2
✓ Branch 1 taken 1274 times.
✗ Branch 2 not taken.
 1274 cache_mgr_->CtrlTxn(object.label, 0, txn);
151
152
1/2
✓ Branch 3 taken 1274 times.
✗ Branch 4 not taken.
 1274 LogCvmfs(kLogCache, kLogDebug, "miss: %s %s", object.label.path.c_str(),
153 url.c_str());
154 1274 TransactionSink sink(cache_mgr_, txn);
155 1274 tls->download_job.SetUrl(&url);
156 1274 tls->download_job.SetSink(&sink);
157 1274 tls->download_job.SetExpectedHash(&object.id);
158 1274 tls->download_job.SetPathInfo(&object.label.path);
159
1/2
✓ Branch 1 taken 1274 times.
✗ Branch 2 not taken.
 1274 ClientCtx *ctx = ClientCtx::GetInstance();
160
3/4
✓ Branch 1 taken 1274 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 140 times.
✓ Branch 4 taken 1134 times.
 1274 if (ctx->IsSet()) {
161
1/2
✓ Branch 5 taken 140 times.
✗ Branch 6 not taken.
 140 ctx->Get(tls->download_job.GetUidPtr(),
162 tls->download_job.GetGidPtr(),
163 tls->download_job.GetPidPtr(),
164 tls->download_job.GetInterruptCuePtr());
165 } else {
166 1134 *(tls->download_job.GetUidPtr()) = -1;
167 1134 *(tls->download_job.GetGidPtr()) = -1;
168 1134 *(tls->download_job.GetPidPtr()) = -1;
169 1134 *(tls->download_job.GetInterruptCuePtr()) = NULL;
170 }
171 1274 tls->download_job.SetCompressed(object.label.zip_algorithm
172 == zlib::kZlibDefault);
173 1274 tls->download_job.SetRangeOffset(object.label.range_offset);
174 1274 tls->download_job.SetRangeSize(static_cast<int64_t>(object.label.size));
175
1/2
✓ Branch 1 taken 1274 times.
✗ Branch 2 not taken.
 1274 download_mgr_->Fetch(&tls->download_job);
176
177
2/2
✓ Branch 1 taken 1104 times.
✓ Branch 2 taken 170 times.
 1274 if (tls->download_job.error_code() == download::kFailOk) {
178
1/2
✓ Branch 2 taken 1104 times.
✗ Branch 3 not taken.
 1104 LogCvmfs(kLogCache, kLogDebug, "finished downloading of %s", url.c_str());
179
180
1/2
✓ Branch 1 taken 1104 times.
✗ Branch 2 not taken.
 1104 fd_return = cache_mgr_->OpenFromTxn(txn);
181
2/2
✓ Branch 0 taken 34 times.
✓ Branch 1 taken 1070 times.
 1104 if (fd_return < 0) {
182
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 cache_mgr_->AbortTxn(txn);
183
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 SignalWaitingThreads(fd_return, object.id, tls);
184 34 return fd_return;
185 }
186
187
1/2
✓ Branch 1 taken 1070 times.
✗ Branch 2 not taken.
 1070 retval = cache_mgr_->CommitTxn(txn);
188
2/2
✓ Branch 0 taken 68 times.
✓ Branch 1 taken 1002 times.
 1070 if (retval < 0) {
189
1/2
✓ Branch 1 taken 68 times.
✗ Branch 2 not taken.
 68 cache_mgr_->Close(fd_return);
190
1/2
✓ Branch 1 taken 68 times.
✗ Branch 2 not taken.
 68 SignalWaitingThreads(retval, object.id, tls);
191 68 return retval;
192 }
193
1/2
✓ Branch 1 taken 1002 times.
✗ Branch 2 not taken.
 1002 SignalWaitingThreads(fd_return, object.id, tls);
194 1002 return fd_return;
195 }
196
197 // Download failed
198
1/5
✗ Branch 4 not taken.
✓ Branch 5 taken 170 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✗ Branch 10 not taken.
 340 LogCvmfs(kLogCache, kLogDebug | kLogSyslogErr,
199 "failed to fetch %s (hash: %s, error %d [%s])",
200
1/2
✓ Branch 1 taken 170 times.
✗ Branch 2 not taken.
 340 object.label.path.c_str(), object.id.ToString().c_str(),
201 170 tls->download_job.error_code(),
202 download::Code2Ascii(tls->download_job.error_code()));
203
1/2
✓ Branch 1 taken 170 times.
✗ Branch 2 not taken.
 170 cache_mgr_->AbortTxn(txn);
204
1/2
✓ Branch 1 taken 170 times.
✗ Branch 2 not taken.
 170 backoff_throttle_->Throttle();
205
1/2
✓ Branch 1 taken 170 times.
✗ Branch 2 not taken.
 170 SignalWaitingThreads(-EIO, object.id, tls);
206 170 return -EIO;
207 1308 }
208
209
210 2456 Fetcher::Fetcher(CacheManager *cache_mgr,
211 download::DownloadManager *download_mgr,
212 BackoffThrottle *backoff_throttle,
213 2456 perf::StatisticsTemplate statistics)
214 2456 : lock_queues_download_(NULL)
215 2456 , lock_tls_blocks_(NULL)
216 2456 , cache_mgr_(cache_mgr)
217 2456 , download_mgr_(download_mgr)
218 2456 , backoff_throttle_(backoff_throttle) {
219 int retval;
220 2456 retval = pthread_key_create(&thread_local_storage_, TLSDestructor);
221
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2456 times.
 2456 assert(retval == 0);
222 2456 lock_queues_download_ = reinterpret_cast<pthread_mutex_t *>(
223 2456 smalloc(sizeof(pthread_mutex_t)));
224 2456 retval = pthread_mutex_init(lock_queues_download_, NULL);
225
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2456 times.
 2456 assert(retval == 0);
226 2456 lock_tls_blocks_ = reinterpret_cast<pthread_mutex_t *>(
227 2456 smalloc(sizeof(pthread_mutex_t)));
228 2456 retval = pthread_mutex_init(lock_tls_blocks_, NULL);
229
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2456 times.
 2456 assert(retval == 0);
230
3/6
✓ Branch 2 taken 2456 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 2456 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 2456 times.
✗ Branch 10 not taken.
 2456 n_downloads = statistics.RegisterTemplated(
231 "n_downloads",
232 "overall number of downloaded files (incl. catalogs, chunks)");
233
3/6
✓ Branch 2 taken 2456 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 2456 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 2456 times.
✗ Branch 10 not taken.
 2456 n_invocations = statistics.RegisterTemplated(
234 "n_invocations",
235 "overall number of object requests (incl. catalogs, chunks)");
236 2456 }
237
238
239 2456 Fetcher::~Fetcher() {
240 int retval;
241
242 {
243 2456 const MutexLockGuard m(lock_tls_blocks_);
244
2/2
✓ Branch 1 taken 895 times.
✓ Branch 2 taken 2456 times.
 3351 for (unsigned i = 0; i < tls_blocks_.size(); ++i)
245 895 CleanupTls(tls_blocks_[i]);
246 2456 }
247
248 2456 retval = pthread_mutex_destroy(lock_tls_blocks_);
249
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2456 times.
 2456 assert(retval == 0);
250 2456 free(lock_tls_blocks_);
251
252 2456 retval = pthread_mutex_destroy(lock_queues_download_);
253
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2456 times.
 2456 assert(retval == 0);
254 2456 free(lock_queues_download_);
255
256 2456 retval = pthread_key_delete(thread_local_storage_);
257
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2456 times.
 2456 assert(retval == 0);
258 2456 }
259
260
261 /**
262 * Depending on the object type, uses either Open() or OpenPinned() from the
263 * cache manager
264 */
265 3304 int Fetcher::OpenSelect(const CacheManager::LabeledObject &object) {
266
5/6
✓ Branch 1 taken 917 times.
✓ Branch 2 taken 2387 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 917 times.
✓ Branch 6 taken 2387 times.
✓ Branch 7 taken 917 times.
 3304 if (object.label.IsCatalog() || object.label.IsPinned()) {
267 2387 return cache_mgr_->OpenPinned(object);
268 } else {
269 917 return cache_mgr_->Open(object);
270 }
271 }
272
273
274 1410 void Fetcher::SignalWaitingThreads(const int fd,
275 const shash::Any &id,
276 ThreadLocalStorage *tls) {
277 1410 const MutexLockGuard m(lock_queues_download_);
278
2/2
✓ Branch 1 taken 136 times.
✓ Branch 2 taken 1410 times.
 1546 for (unsigned i = 0, s = tls->other_pipes_waiting.size(); i < s; ++i) {
279
3/4
✓ Branch 0 taken 102 times.
✓ Branch 1 taken 34 times.
✓ Branch 3 taken 102 times.
✗ Branch 4 not taken.
 136 int fd_dup = (fd >= 0) ? cache_mgr_->Dup(fd) : fd;
280
1/2
✓ Branch 2 taken 136 times.
✗ Branch 3 not taken.
 136 WritePipe(tls->other_pipes_waiting[i], &fd_dup, sizeof(int));
281 }
282 1410 tls->other_pipes_waiting.clear();
283
1/2
✓ Branch 1 taken 1410 times.
✗ Branch 2 not taken.
 1410 queues_download_.erase(id);
284 1410 }
285
286 } // namespace cvmfs
287