GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/fetch.cc
Date: 2026-07-19 02:35:15
Exec Total Coverage
Lines: 158 168 94.0%
Branches: 105 195 53.8%
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 88 void TLSDestructor(void *data) {
24 88 Fetcher::ThreadLocalStorage *tls = static_cast<Fetcher::ThreadLocalStorage *>(
25 data);
26 88 std::vector<Fetcher::ThreadLocalStorage *> *tls_blocks = &tls->fetcher
27 ->tls_blocks_;
28
29 {
30 88 const MutexLockGuard m(tls->fetcher->lock_tls_blocks_);
31 88 for (vector<Fetcher::ThreadLocalStorage *>::iterator
32 88 i = tls_blocks->begin(),
33 88 iEnd = tls_blocks->end();
34
1/2
✓ Branch 1 taken 176 times.
✗ Branch 2 not taken.
 176 i != iEnd;
35 88 ++i) {
36
2/2
✓ Branch 1 taken 88 times.
✓ Branch 2 taken 88 times.
 176 if ((*i) == tls) {
37
1/2
✓ Branch 2 taken 88 times.
✗ Branch 3 not taken.
 88 tls_blocks->erase(i);
38 88 break;
39 }
40 }
41 88 }
42 88 tls->fetcher->CleanupTls(tls);
43 88 }
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 1012 void Fetcher::CleanupTls(ThreadLocalStorage *tls) {
51 1012 ClosePipe(tls->pipe_wait);
52
1/2
✓ Branch 0 taken 1012 times.
✗ Branch 1 not taken.
 1012 delete tls;
53 1012 }
54
55
56 /**
57 * Initialized thread-local storage if called the first time in a new thread.
58 */
59 2014 Fetcher::ThreadLocalStorage *Fetcher::GetTls() {
60 ThreadLocalStorage *tls = static_cast<ThreadLocalStorage *>(
61 2014 pthread_getspecific(thread_local_storage_));
62
2/2
✓ Branch 0 taken 1002 times.
✓ Branch 1 taken 1012 times.
 2014 if (tls != NULL)
63 1002 return tls;
64
65
2/4
✓ Branch 1 taken 1012 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 1012 times.
✗ Branch 5 not taken.
 1012 tls = new ThreadLocalStorage();
66 1012 tls->fetcher = this;
67
1/2
✓ Branch 1 taken 1012 times.
✗ Branch 2 not taken.
 1012 MakePipe(tls->pipe_wait);
68 1012 tls->download_job.SetCompressed(true);
69 1012 tls->download_job.SetProbeHosts(true);
70 1012 const int retval = pthread_setspecific(thread_local_storage_, tls);
71
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1012 times.
 1012 assert(retval == 0);
72
73 1012 const MutexLockGuard m(lock_tls_blocks_);
74
1/2
✓ Branch 1 taken 1012 times.
✗ Branch 2 not taken.
 1012 tls_blocks_.push_back(tls);
75
76 1012 return tls;
77 1012 }
78
79
80 2094 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 2094 perf::Inc(n_invocations);
86
87 // Try to open from local cache
88
3/4
✓ Branch 1 taken 2094 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 520 times.
✓ Branch 4 taken 1574 times.
 2094 if ((fd_return = OpenSelect(object)) >= 0) {
89
1/2
✓ Branch 2 taken 520 times.
✗ Branch 3 not taken.
 520 LogCvmfs(kLogCache, kLogDebug, "hit: %s", object.label.path.c_str());
90 520 return fd_return;
91 }
92
93
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 1574 times.
 1574 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 1574 times.
✗ Branch 2 not taken.
 1574 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 1574 pthread_mutex_lock(lock_queues_download_);
106 1574 const ThreadQueues::iterator iDownloadQueue = queues_download_.find(
107
1/2
✓ Branch 1 taken 1574 times.
✗ Branch 2 not taken.
 1574 object.id);
108
2/2
✓ Branch 2 taken 44 times.
✓ Branch 3 taken 1530 times.
 1574 if (iDownloadQueue != queues_download_.end()) {
109
1/2
✓ Branch 2 taken 44 times.
✗ Branch 3 not taken.
 44 LogCvmfs(kLogCache, kLogDebug, "waiting for download of %s",
110 object.label.path.c_str());
111
112
1/2
✓ Branch 2 taken 44 times.
✗ Branch 3 not taken.
 44 iDownloadQueue->second->push_back(tls->pipe_wait[1]);
113 44 pthread_mutex_unlock(lock_queues_download_);
114
1/2
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
 44 ReadPipe(tls->pipe_wait[0], &fd_return, sizeof(int));
115
116
1/2
✓ Branch 2 taken 44 times.
✗ Branch 3 not taken.
 44 LogCvmfs(kLogCache, kLogDebug, "received from another thread fd %d for %s",
117 fd_return, object.label.path.c_str());
118 44 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 1530 times.
✗ Branch 2 not taken.
 1530 fd_return = OpenSelect(object);
122
2/2
✓ Branch 0 taken 44 times.
✓ Branch 1 taken 1486 times.
 1530 if (fd_return >= 0) {
123 44 pthread_mutex_unlock(lock_queues_download_);
124 44 return fd_return;
125 }
126
127 // Create a new queue for this chunk
128
1/2
✓ Branch 1 taken 1486 times.
✗ Branch 2 not taken.
 1486 queues_download_[object.id] = &tls->other_pipes_waiting;
129 1486 pthread_mutex_unlock(lock_queues_download_);
130 }
131
132 1486 perf::Inc(n_downloads);
133
134 // Involve the download manager
135
1/2
✓ Branch 2 taken 1486 times.
✗ Branch 3 not taken.
 1486 LogCvmfs(kLogCache, kLogDebug, "downloading %s", object.label.path.c_str());
136 1486 std::string url;
137
2/2
✓ Branch 1 taken 132 times.
✓ Branch 2 taken 1354 times.
 1486 if (object.label.IsExternal()) {
138
2/4
✓ Branch 1 taken 132 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 132 times.
✗ Branch 5 not taken.
 132 url = !alt_url.empty() ? alt_url : object.label.path;
139 } else {
140
8/14
✓ Branch 1 taken 44 times.
✓ Branch 2 taken 1310 times.
✓ Branch 4 taken 44 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 1310 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 1310 times.
✗ Branch 11 not taken.
✓ Branch 13 taken 1354 times.
✗ Branch 14 not taken.
✓ Branch 18 taken 1310 times.
✓ Branch 19 taken 44 times.
✗ Branch 22 not taken.
✗ Branch 23 not taken.
 1354 url = "/" + (alt_url.size() ? alt_url : "data/" + object.id.MakePath());
141 }
142
1/2
✓ Branch 1 taken 1486 times.
✗ Branch 2 not taken.
 1486 void *txn = alloca(cache_mgr_->SizeOfTxn());
143
1/2
✓ Branch 1 taken 1486 times.
✗ Branch 2 not taken.
 1486 retval = cache_mgr_->StartTxn(object.id, object.label.size, txn);
144
2/2
✓ Branch 0 taken 44 times.
✓ Branch 1 taken 1442 times.
 1486 if (retval < 0) {
145
1/2
✓ Branch 2 taken 44 times.
✗ Branch 3 not taken.
 44 LogCvmfs(kLogCache, kLogDebug, "could not start transaction on %s",
146 object.label.path.c_str());
147
1/2
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
 44 SignalWaitingThreads(retval, object.id, tls);
148 44 return retval;
149 }
150
1/2
✓ Branch 1 taken 1442 times.
✗ Branch 2 not taken.
 1442 cache_mgr_->CtrlTxn(object.label, 0, txn);
151
152
1/2
✓ Branch 3 taken 1442 times.
✗ Branch 4 not taken.
 1442 LogCvmfs(kLogCache, kLogDebug, "miss: %s %s", object.label.path.c_str(),
153 url.c_str());
154 1442 TransactionSink sink(cache_mgr_, txn);
155 1442 tls->download_job.SetUrl(&url);
156 1442 tls->download_job.SetSink(&sink);
157 1442 tls->download_job.SetExpectedHash(&object.id);
158 1442 tls->download_job.SetPathInfo(&object.label.path);
159
1/2
✓ Branch 1 taken 1442 times.
✗ Branch 2 not taken.
 1442 ClientCtx *ctx = ClientCtx::GetInstance();
160
3/4
✓ Branch 1 taken 1442 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 194 times.
✓ Branch 4 taken 1248 times.
 1442 if (ctx->IsSet()) {
161
1/2
✓ Branch 5 taken 194 times.
✗ Branch 6 not taken.
 194 ctx->Get(tls->download_job.GetUidPtr(),
162 tls->download_job.GetGidPtr(),
163 tls->download_job.GetPidPtr(),
164 tls->download_job.GetInterruptCuePtr());
165 } else {
166 1248 *(tls->download_job.GetUidPtr()) = -1;
167 1248 *(tls->download_job.GetGidPtr()) = -1;
168 1248 *(tls->download_job.GetPidPtr()) = -1;
169 1248 *(tls->download_job.GetInterruptCuePtr()) = NULL;
170 }
171 1442 tls->download_job.SetCompressed(object.label.zip_algorithm
172 == zlib::kZlibDefault);
173 1442 tls->download_job.SetRangeOffset(object.label.range_offset);
174 1442 tls->download_job.SetRangeSize(static_cast<int64_t>(object.label.size));
175
1/2
✓ Branch 1 taken 1442 times.
✗ Branch 2 not taken.
 1442 download_mgr_->Fetch(&tls->download_job);
176
177 // Partial replica failover: a partial Stratum-1 serves a 404 for objects it
178 // did not replicate. Retry once against the full replica and let the common
179 // success/error paths below handle the result. Only 404 triggers failover:
180 // other HTTP errors (403, 5xx, ...) also map to kFailHostHttp but indicate a
181 // server problem rather than an unreplicated object, so they must not be
182 // masked by silently fetching from elsewhere. The txn buffer is reused in
183 // place by StartTxn, so `sink` keeps pointing at the fresh transaction and
184 // needs no reconstruction.
185 1442 if (tls->download_job.error_code() == download::kFailHostHttp
186
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 44 times.
 44 && tls->download_job.http_code() == 404
187
3/6
✓ Branch 0 taken 44 times.
✓ Branch 1 taken 1398 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 1442 times.
 1486 && full_replica_download_mgr_ != NULL) {
188 ✗ LogCvmfs(kLogCache, kLogDebug | kLogSyslog,
189 "partial replica: %s not available on primary, "
190 "retrying from full replica",
191 object.label.path.c_str());
192 ✗ cache_mgr_->AbortTxn(txn);
193 ✗ retval = cache_mgr_->StartTxn(object.id, object.label.size, txn);
194 ✗ if (retval < 0) {
195 ✗ SignalWaitingThreads(retval, object.id, tls);
196 ✗ return retval;
197 }
198 ✗ cache_mgr_->CtrlTxn(object.label, 0, txn);
199 ✗ full_replica_download_mgr_->Fetch(&tls->download_job);
200 }
201
202
2/2
✓ Branch 1 taken 1217 times.
✓ Branch 2 taken 225 times.
 1442 if (tls->download_job.error_code() == download::kFailOk) {
203
1/2
✓ Branch 2 taken 1217 times.
✗ Branch 3 not taken.
 1217 LogCvmfs(kLogCache, kLogDebug, "finished downloading of %s", url.c_str());
204
205
1/2
✓ Branch 1 taken 1217 times.
✗ Branch 2 not taken.
 1217 fd_return = cache_mgr_->OpenFromTxn(txn);
206
2/2
✓ Branch 0 taken 44 times.
✓ Branch 1 taken 1173 times.
 1217 if (fd_return < 0) {
207
1/2
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
 44 cache_mgr_->AbortTxn(txn);
208
1/2
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
 44 SignalWaitingThreads(fd_return, object.id, tls);
209 44 return fd_return;
210 }
211
212
1/2
✓ Branch 1 taken 1173 times.
✗ Branch 2 not taken.
 1173 retval = cache_mgr_->CommitTxn(txn);
213
2/2
✓ Branch 0 taken 88 times.
✓ Branch 1 taken 1085 times.
 1173 if (retval < 0) {
214
1/2
✓ Branch 1 taken 88 times.
✗ Branch 2 not taken.
 88 cache_mgr_->Close(fd_return);
215
1/2
✓ Branch 1 taken 88 times.
✗ Branch 2 not taken.
 88 SignalWaitingThreads(retval, object.id, tls);
216 88 return retval;
217 }
218
1/2
✓ Branch 1 taken 1085 times.
✗ Branch 2 not taken.
 1085 SignalWaitingThreads(fd_return, object.id, tls);
219 1085 return fd_return;
220 }
221
222 // Download failed (primary, and full replica if configured)
223
1/5
✗ Branch 4 not taken.
✓ Branch 5 taken 225 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✗ Branch 10 not taken.
 450 LogCvmfs(kLogCache, kLogDebug | kLogSyslogErr,
224 "failed to fetch %s (hash: %s, error %d [%s])",
225
1/2
✓ Branch 1 taken 225 times.
✗ Branch 2 not taken.
 450 object.label.path.c_str(), object.id.ToString().c_str(),
226 225 tls->download_job.error_code(),
227 download::Code2Ascii(tls->download_job.error_code()));
228
1/2
✓ Branch 1 taken 225 times.
✗ Branch 2 not taken.
 225 cache_mgr_->AbortTxn(txn);
229
1/2
✓ Branch 1 taken 225 times.
✗ Branch 2 not taken.
 225 backoff_throttle_->Throttle();
230
1/2
✓ Branch 1 taken 225 times.
✗ Branch 2 not taken.
 225 SignalWaitingThreads(-EIO, object.id, tls);
231 225 return -EIO;
232 1486 }
233
234
235 2358 Fetcher::Fetcher(CacheManager *cache_mgr,
236 download::DownloadManager *download_mgr,
237 BackoffThrottle *backoff_throttle,
238 2358 perf::StatisticsTemplate statistics)
239 2358 : lock_queues_download_(NULL)
240 2358 , lock_tls_blocks_(NULL)
241 2358 , cache_mgr_(cache_mgr)
242 2358 , download_mgr_(download_mgr)
243 2358 , full_replica_download_mgr_(NULL)
244 2358 , backoff_throttle_(backoff_throttle) {
245 int retval;
246 2358 retval = pthread_key_create(&thread_local_storage_, TLSDestructor);
247
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2358 times.
 2358 assert(retval == 0);
248 2358 lock_queues_download_ = reinterpret_cast<pthread_mutex_t *>(
249 2358 smalloc(sizeof(pthread_mutex_t)));
250 2358 retval = pthread_mutex_init(lock_queues_download_, NULL);
251
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2358 times.
 2358 assert(retval == 0);
252 2358 lock_tls_blocks_ = reinterpret_cast<pthread_mutex_t *>(
253 2358 smalloc(sizeof(pthread_mutex_t)));
254 2358 retval = pthread_mutex_init(lock_tls_blocks_, NULL);
255
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2358 times.
 2358 assert(retval == 0);
256
3/6
✓ Branch 2 taken 2358 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 2358 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 2358 times.
✗ Branch 10 not taken.
 2358 n_downloads = statistics.RegisterTemplated(
257 "n_downloads",
258 "overall number of downloaded files (incl. catalogs, chunks)");
259
3/6
✓ Branch 2 taken 2358 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 2358 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 2358 times.
✗ Branch 10 not taken.
 2358 n_invocations = statistics.RegisterTemplated(
260 "n_invocations",
261 "overall number of object requests (incl. catalogs, chunks)");
262 2358 }
263
264
265 8668 Fetcher::~Fetcher() {
266 int retval;
267
268 {
269 4716 const MutexLockGuard m(lock_tls_blocks_);
270
2/2
✓ Branch 1 taken 924 times.
✓ Branch 2 taken 2358 times.
 6564 for (unsigned i = 0; i < tls_blocks_.size(); ++i)
271 1848 CleanupTls(tls_blocks_[i]);
272 }
273
274 4716 retval = pthread_mutex_destroy(lock_tls_blocks_);
275
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2358 times.
 4716 assert(retval == 0);
276 4716 free(lock_tls_blocks_);
277
278 4716 retval = pthread_mutex_destroy(lock_queues_download_);
279
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2358 times.
 4716 assert(retval == 0);
280 4716 free(lock_queues_download_);
281
282 4716 retval = pthread_key_delete(thread_local_storage_);
283
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2358 times.
 4716 assert(retval == 0);
284 8668 }
285
286
287 /**
288 * Depending on the object type, uses either Open() or OpenPinned() from the
289 * cache manager
290 */
291 3624 int Fetcher::OpenSelect(const CacheManager::LabeledObject &object) {
292
5/6
✓ Branch 1 taken 1037 times.
✓ Branch 2 taken 2587 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 1037 times.
✓ Branch 6 taken 2587 times.
✓ Branch 7 taken 1037 times.
 3624 if (object.label.IsCatalog() || object.label.IsPinned()) {
293 2587 return cache_mgr_->OpenPinned(object);
294 } else {
295 1037 return cache_mgr_->Open(object);
296 }
297 }
298
299
300 1618 void Fetcher::SignalWaitingThreads(const int fd,
301 const shash::Any &id,
302 ThreadLocalStorage *tls) {
303 1618 const MutexLockGuard m(lock_queues_download_);
304
2/2
✓ Branch 1 taken 176 times.
✓ Branch 2 taken 1618 times.
 1794 for (unsigned i = 0, s = tls->other_pipes_waiting.size(); i < s; ++i) {
305
3/4
✓ Branch 0 taken 132 times.
✓ Branch 1 taken 44 times.
✓ Branch 3 taken 132 times.
✗ Branch 4 not taken.
 176 int fd_dup = (fd >= 0) ? cache_mgr_->Dup(fd) : fd;
306
1/2
✓ Branch 2 taken 176 times.
✗ Branch 3 not taken.
 176 WritePipe(tls->other_pipes_waiting[i], &fd_dup, sizeof(int));
307 }
308 1618 tls->other_pipes_waiting.clear();
309
1/2
✓ Branch 1 taken 1618 times.
✗ Branch 2 not taken.
 1618 queues_download_.erase(id);
310 1618 }
311
312 } // namespace cvmfs
313