GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/bundle_mgr.cc
Date: 2026-07-26 02:35:14
Exec Total Coverage
Lines: 147 212 69.3%
Branches: 113 304 37.2%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #include "bundle_mgr.h"
6
7 #include <fcntl.h>
8 #include <pthread.h>
9 #include <unistd.h>
10
11 #include <cassert>
12 #include <cerrno>
13 #include <cstdlib>
14 #include <memory>
15 #include <string>
16 #include <vector>
17
18 #include "cache.h"
19 #include "catalog_mgr_client.h"
20 #include "fetch.h"
21 #include "file_chunk.h"
22 #include "json_document.h"
23 #include "mountpoint.h"
24 #include "options.h"
25 #include "shortstring.h"
26 #include "util/posix.h"
27
28 namespace {
29 constexpr size_t kDefaultBundlePoolSize = 8;
30
31 // Read the .cvmfsbundle-<basename> file via the cvmfs cache
32 190 BundleFileMgr *LoadBundleFromCvmfs(MountPoint *mp,
33 const PathString &bundle_file_path) {
34
1/2
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
 190 catalog::DirectoryEntry dirent;
35
2/4
✓ Branch 2 taken 190 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 190 times.
 190 if (!mp->catalog_mgr()->LookupPath(bundle_file_path, catalog::kLookupDefault,
36 &dirent)) {
37 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: LookupPath failed for %s",
38 ✗ bundle_file_path.ToString().c_str());
39 ✗ return nullptr;
40 }
41 190 cvmfs::Fetcher *fetcher = mp->fetcher();
42
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 190 times.
 190 if (fetcher == nullptr) {
43 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: fetcher is null");
44 ✗ return nullptr;
45 }
46
47 190 CacheManager::Label label;
48
1/2
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
 190 label.path = bundle_file_path.ToString();
49
1/2
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
 190 label.size = dirent.size();
50 190 label.zip_algorithm = dirent.compression_algorithm();
51
2/4
✓ Branch 2 taken 190 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 190 times.
✗ Branch 6 not taken.
 570 const int fd = fetcher->Fetch(
52
1/2
✓ Branch 2 taken 190 times.
✗ Branch 3 not taken.
 380 CacheManager::LabeledObject(dirent.checksum(), label));
53
1/2
✓ Branch 0 taken 190 times.
✗ Branch 1 not taken.
 190 if (fd < 0) {
54
1/2
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
 190 LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: Fetch returned fd=%d", fd);
55 190 return nullptr;
56 }
57
58 ✗ CacheManager *cache_mgr = mp->file_system()->cache_mgr();
59 ✗ std::string content;
60 ✗ content.resize(static_cast<size_t>(dirent.size()));
61 ✗ const int64_t n = cache_mgr->Pread(fd, &content[0], content.size(), 0);
62 ✗ cache_mgr->Close(fd);
63 ✗ if (n < 0 || static_cast<size_t>(n) != content.size()) {
64 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: Pread returned %ld want %zu",
65 static_cast<long>(n), content.size());
66 ✗ return nullptr;
67 }
68
69 // The bundle file may start with a "#%CVMFS_BUNDLE version=..." header
70 // line (per file_bundle.h); strip any leading lines beginning with '#'
71 // before handing off to the strict JSON parser.
72 ✗ size_t json_start = 0;
73 ✗ while (json_start < content.size() && content[json_start] == '#') {
74 ✗ const size_t nl = content.find('\n', json_start);
75 ✗ if (nl == std::string::npos) {
76 ✗ json_start = content.size();
77 ✗ break;
78 }
79 ✗ json_start = nl + 1;
80 }
81 const std::string json_text = (json_start == 0) ? content
82 ✗ : content.substr(json_start);
83
84 ✗ JsonDocument *doc = JsonDocument::Create(json_text);
85 ✗ if (doc == nullptr) {
86 ✗ LogCvmfs(kLogCvmfs, kLogDebug,
87 "BUNDLE-LOAD: JsonDocument::Create failed (size=%zu)",
88 json_text.size());
89 ✗ return nullptr;
90 }
91 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: loaded bundle %s (%zu bytes)",
92 ✗ bundle_file_path.ToString().c_str(), content.size());
93 ✗ return new BundleFileMgr(doc);
94 190 }
95 } // namespace
96
97 190 BundleMgr::BundleMgr(MountPoint *mp, const PathString &path)
98 190 : mount_point_(mp)
99 190 , path_(path)
100 190 , fetcher_threads_()
101 190 , pool_size_(kDefaultBundlePoolSize) {
102
2/4
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 190 times.
✗ Branch 5 not taken.
 190 fname_ = GetFileName(path_);
103
2/4
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 190 times.
✗ Branch 5 not taken.
 190 parent_path_ = GetParentPath(path_);
104 // There is a naming convention regarding the name of the file with the
105 // contents of the bundle
106
3/6
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 190 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 190 times.
✗ Branch 8 not taken.
 760 bundle_file_path_ = PathString(parent_path_.ToString() + "/.cvmfsbundle-"
107
3/6
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 190 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 190 times.
✗ Branch 8 not taken.
 950 + fname_.ToString());
108
109 190 pipe_bm_[0] = pipe_bm_[1] = -1;
110 190 pthread_mutex_init(&worker_read_mutex_, nullptr);
111
112
1/2
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
 190 bfm_ = LoadBundleFromCvmfs(mount_point_, bundle_file_path_);
113
1/2
✓ Branch 0 taken 190 times.
✗ Branch 1 not taken.
 190 if (bfm_ == nullptr) {
114
1/2
✓ Branch 2 taken 190 times.
✗ Branch 3 not taken.
 190 LogCvmfs(kLogCvmfs, kLogDebug, "BundleMgr: failed to load bundle file %s",
115
1/2
✓ Branch 1 taken 190 times.
✗ Branch 2 not taken.
 380 bundle_file_path_.ToString().c_str());
116 190 is_valid_ = false;
117 190 return;
118 }
119
120 // Pool size override via CVMFS_BUNDLE_POOL_SIZE
121 ✗ if (mount_point_ != nullptr && mount_point_->file_system() != nullptr
122 ✗ && mount_point_->file_system()->options_mgr() != nullptr) {
123 ✗ std::string opt;
124 ✗ if (mount_point_->file_system()->options_mgr()->GetValue(
125 "CVMFS_BUNDLE_POOL_SIZE", &opt)) {
126 ✗ char *end = nullptr;
127 ✗ const unsigned long n = std::strtoul(opt.c_str(), &end, 10);
128 ✗ if (end != opt.c_str() && n >= 1) {
129 ✗ pool_size_ = static_cast<size_t>(n);
130 }
131 }
132 }
133
134 ✗ SpawnFetcherPool();
135 }
136
137 76 void BundleMgr::Fetch() {
138
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 76 times.
 76 if (not is_valid_) {
139 ✗ LogCvmfs(kLogBundleMgr,
140 kLogDebug,
141 "BundleMgr is not in a valid state. Can't fetch!");
142 ✗ return;
143 }
144
145
4/6
✓ Branch 1 taken 304 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 304 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 76 times.
✓ Branch 7 taken 228 times.
 304 while (auto file = bfm_->GetNext()) {
146
1/2
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
 228 const PathString path = NormalizeDependencyPath(file);
147 // A TrySendPath() here is used as a profylaxis to a scenario where the pipe
148 // is currently blocked.
149
2/4
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 228 times.
 228 while (not TrySendPath(back_channel_, path)) {
150 }
151
2/2
✓ Branch 2 taken 228 times.
✓ Branch 3 taken 76 times.
 532 }
152 }
153
154 /**
155 * Dependency paths in a bundle are absolute from the repository root.
156 * Entries without a leading slash (optionally prefixed with "./") are
157 * resolved relative to the directory holding the bundle file.
158 */
159 228 PathString BundleMgr::NormalizeDependencyPath(const PathString &path) const {
160
2/4
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 228 times.
 228 if (path.StartsWith(PathString("/", 1)))
161 ✗ return path;
162
1/2
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
 228 const PathString relative = path.StartsWith(PathString("./", 2))
163 ? path.Suffix(2)
164
2/6
✗ Branch 0 not taken.
✓ Branch 1 taken 228 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 6 taken 228 times.
✗ Branch 7 not taken.
 228 : path;
165
1/2
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
 228 PathString normalized(parent_path_);
166
1/2
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
 228 normalized.Append("/", 1);
167
1/2
✓ Branch 3 taken 228 times.
✗ Branch 4 not taken.
 228 normalized.Append(relative.GetChars(), relative.GetLength());
168
1/2
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
 228 return normalized;
169 228 }
170
171 266 void BundleMgr::JoinFetcherPool() {
172
2/2
✓ Branch 0 taken 76 times.
✓ Branch 1 taken 190 times.
 266 if (pipe_bm_[1] < 0)
173 76 return;
174 // Send one kTerminate per worker. Workers drain all queued kFetch
175 // messages before reaching their kTerminate (FIFO pipe), so we can't
176 // just close the pipe — that would EOF some workers mid-drain.
177
2/2
✓ Branch 1 taken 1520 times.
✓ Branch 2 taken 190 times.
 1710 for (size_t i = 0; i < fetcher_threads_.size(); ++i) {
178 1520 Command cmd = Command::kTerminate;
179 while (true) {
180
1/2
✓ Branch 1 taken 1520 times.
✗ Branch 2 not taken.
 1520 const ssize_t n = ::write(pipe_bm_[1], &cmd, sizeof(Command));
181
1/2
✓ Branch 0 taken 1520 times.
✗ Branch 1 not taken.
 1520 if (n == sizeof(Command))
182 1520 break;
183 ✗ if (errno != EAGAIN && errno != EWOULDBLOCK)
184 ✗ break;
185 }
186 }
187 // Wait for every worker to drain its share of the queue and exit.
188
2/2
✓ Branch 5 taken 1520 times.
✓ Branch 6 taken 190 times.
 1710 for (auto &t : fetcher_threads_) {
189
1/2
✓ Branch 2 taken 1520 times.
✗ Branch 3 not taken.
 1520 pthread_join(*t, nullptr);
190 }
191 190 fetcher_threads_.clear();
192 190 ClosePipe(pipe_bm_);
193 // Mark the pool as gone so that a subsequent call (e.g. the destructor
194 // running after an explicit JoinFetcherPool()) is a no-op instead of a
195 // double close/join.
196 190 pipe_bm_[0] = pipe_bm_[1] = -1;
197 }
198
199 190 void BundleMgr::SpawnFetcherPool() {
200 190 MakePipe(pipe_bm_);
201 190 back_channel_ = pipe_bm_[1];
202
203 // Non-blocking writes on the work-queue pipe so TrySendPath can poll.
204 // Per pipe(7), writes <= PIPE_BUF are atomic on non-blocking pipes:
205 // they either fully succeed or fail with EAGAIN.
206 190 const int flags = fcntl(back_channel_, F_GETFL);
207 190 fcntl(back_channel_, F_SETFL, flags | O_NONBLOCK);
208
209
2/2
✓ Branch 0 taken 1520 times.
✓ Branch 1 taken 190 times.
 1710 for (size_t i = 0; i < pool_size_; ++i) {
210
1/2
✓ Branch 1 taken 1520 times.
✗ Branch 2 not taken.
 1520 std::unique_ptr<pthread_t> thread(new pthread_t());
211 1520 const int res = pthread_create(thread.get(), nullptr, MainBundleMgrFetcher,
212 this);
213
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1520 times.
 1520 if (res != 0) {
214 ✗ LogCvmfs(kLogBundleMgr, kLogDebug,
215 "Thread creation failed! pool_size_=%zu spawned=%zu", pool_size_,
216 i);
217 ✗ is_valid_ = false;
218 ✗ return;
219 }
220
1/2
✓ Branch 2 taken 1520 times.
✗ Branch 3 not taken.
 1520 fetcher_threads_.emplace_back(std::move(thread));
221
1/2
✓ Branch 1 taken 1520 times.
✗ Branch 2 not taken.
 1520 }
222 }
223
224 228 void BundleMgr::FetchPath(const PathString &path) {
225
1/2
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
 228 catalog::DirectoryEntry dirent;
226
1/2
✓ Branch 2 taken 228 times.
✗ Branch 3 not taken.
 228 const bool found = mount_point_->catalog_mgr()->LookupPath(
227 path, catalog::kLookupDefault, &dirent);
228 228 cvmfs::Fetcher *this_fetcher = dirent.IsExternalFile()
229
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 228 times.
 228 ? mount_point_->external_fetcher()
230 228 : mount_point_->fetcher();
231
2/4
✓ Branch 0 taken 228 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 228 times.
 228 if (not(found and this_fetcher)) {
232 ✗ LogCvmfs(kLogCvmfs, kLogDebug,
233 "BUNDLE-FETCH: lookup failed for %s (found=%d)",
234 ✗ path.ToString().c_str(), int(found));
235 ✗ return;
236 }
237
1/2
✓ Branch 2 taken 228 times.
✗ Branch 3 not taken.
 228 LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-FETCH: prefetching %s",
238
1/2
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
 456 path.ToString().c_str());
239
240
2/2
✓ Branch 1 taken 114 times.
✓ Branch 2 taken 114 times.
 228 if (dirent.IsChunkedFile()) {
241 // Files above the chunking threshold are stored as per-chunk objects;
242 // their bulk object only exists if the repository sets
243 // CVMFS_GENERATE_LEGACY_BULK_CHUNKS. Fetch the chunks, exactly like the
244 // read path does.
245
1/2
✓ Branch 1 taken 114 times.
✗ Branch 2 not taken.
 114 FileChunkList chunks;
246
1/2
✓ Branch 3 taken 114 times.
✗ Branch 4 not taken.
 114 if (!mount_point_->catalog_mgr()->ListFileChunks(
247 path, dirent.hash_algorithm(), &chunks)
248
3/6
✓ Branch 0 taken 114 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 114 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 114 times.
 114 || chunks.IsEmpty()) {
249 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-FETCH: no chunks found for %s",
250 ✗ path.ToString().c_str());
251 ✗ return;
252 }
253
2/2
✓ Branch 1 taken 342 times.
✓ Branch 2 taken 114 times.
 456 for (unsigned i = 0; i < chunks.size(); ++i) {
254 342 CacheManager::Label label;
255
1/2
✓ Branch 1 taken 342 times.
✗ Branch 2 not taken.
 342 label.path = path.ToString();
256 342 label.size = chunks.AtPtr(i)->size();
257 342 label.zip_algorithm = dirent.compression_algorithm();
258 342 label.flags |= CacheManager::kLabelChunked;
259
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 342 times.
 342 if (mount_point_->catalog_mgr()->volatile_flag())
260 ✗ label.flags |= CacheManager::kLabelVolatile;
261
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 342 times.
 342 if (dirent.IsExternalFile()) {
262 ✗ label.flags |= CacheManager::kLabelExternal;
263 ✗ label.range_offset = chunks.AtPtr(i)->offset();
264 }
265
2/4
✓ Branch 2 taken 342 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 342 times.
✗ Branch 6 not taken.
 1026 const int fd = this_fetcher->Fetch(
266
1/2
✓ Branch 3 taken 342 times.
✗ Branch 4 not taken.
 684 CacheManager::LabeledObject(chunks.AtPtr(i)->content_hash(), label));
267
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 342 times.
 342 if (fd >= 0)
268 ✗ mount_point_->file_system()->cache_mgr()->Close(fd);
269 342 }
270 114 return;
271 114 }
272
273 114 CacheManager::Label label;
274
1/2
✓ Branch 1 taken 114 times.
✗ Branch 2 not taken.
 114 label.path = path.ToString();
275
1/2
✓ Branch 1 taken 114 times.
✗ Branch 2 not taken.
 114 label.size = dirent.size();
276 114 label.zip_algorithm = dirent.compression_algorithm();
277
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 114 times.
 114 if (mount_point_->catalog_mgr()->volatile_flag())
278 ✗ label.flags |= CacheManager::kLabelVolatile;
279
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 114 times.
 114 if (dirent.IsExternalFile())
280 ✗ label.flags |= CacheManager::kLabelExternal;
281
2/4
✓ Branch 2 taken 114 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 114 times.
✗ Branch 6 not taken.
 342 const int fd = this_fetcher->Fetch(
282
1/2
✓ Branch 2 taken 114 times.
✗ Branch 3 not taken.
 228 CacheManager::LabeledObject(dirent.checksum(), label));
283
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 114 times.
 114 if (fd >= 0)
284 ✗ mount_point_->file_system()->cache_mgr()->Close(fd);
285
2/2
✓ Branch 2 taken 114 times.
✓ Branch 3 taken 114 times.
 228 }
286
287 1520 void *BundleMgr::MainBundleMgrFetcher(void *data) {
288 #ifndef __APPLE__
289 1520 pthread_setname_np(pthread_self(), "bm_fetcher");
290 #endif
291 1520 BundleMgr *mgr = static_cast<BundleMgr *>(data);
292 1520 const int rfd = mgr->pipe_bm_[0];
293
294 while (true) {
295 1748 Command cmd = Command::kTerminate;
296 1748 PathString path;
297 1748 bool got_path = false;
298 1748 bool eof = false;
299
300 // Atomically receive cmd + (optional) path payload. The whole receipt
301 // is under worker_read_mutex_ so messages aren't interleaved between
302 // workers reading from the shared pipe.
303 1748 pthread_mutex_lock(&mgr->worker_read_mutex_);
304
1/2
✓ Branch 1 taken 1748 times.
✗ Branch 2 not taken.
 1748 const ssize_t n = read(rfd, &cmd, sizeof(Command));
305
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1748 times.
 1748 if (n != static_cast<ssize_t>(sizeof(Command))) {
306 ✗ eof = true;
307
2/2
✓ Branch 0 taken 228 times.
✓ Branch 1 taken 1520 times.
 1748 } else if (cmd == Command::kFetch) {
308
2/4
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 228 times.
✗ Branch 5 not taken.
 228 path = mgr->ReceivePath(rfd);
309 228 got_path = true;
310 }
311 1748 pthread_mutex_unlock(&mgr->worker_read_mutex_);
312
313
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1748 times.
 1748 if (eof)
314 ✗ break;
315
316 1748 bool terminate = false;
317
2/2
✓ Branch 0 taken 228 times.
✓ Branch 1 taken 1520 times.
 1748 switch (cmd) {
318 228 case Command::kFetch: {
319
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 228 times.
 228 if (!got_path) {
320 ✗ terminate = true;
321 ✗ break;
322 }
323
1/2
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
 228 mgr->FetchPath(path);
324 228 } break;
325 1520 case Command::kTerminate:
326 default:
327 1520 terminate = true;
328 1520 break;
329 }
330
2/2
✓ Branch 0 taken 1520 times.
✓ Branch 1 taken 228 times.
 1748 if (terminate) {
331 1520 break;
332 }
333
2/2
✓ Branch 1 taken 228 times.
✓ Branch 2 taken 1520 times.
 1976 }
334
335 1520 pthread_exit(nullptr);
336 }
337
338 304 PathString BundleMgr::ReceivePath(int fd) const {
339
1/2
✓ Branch 1 taken 304 times.
✗ Branch 2 not taken.
 304 const std::string buffer = BlockingReceive(fd);
340
1/2
✓ Branch 1 taken 304 times.
✗ Branch 2 not taken.
 304 assert(buffer.size() > 0 && "A path can't be empty");
341
1/2
✓ Branch 1 taken 304 times.
✗ Branch 2 not taken.
 608 return PathString(buffer);
342 304 }
343
344 228 bool BundleMgr::TrySendPath(int fd, const PathString &path) const {
345 228 Command cmd = Command::kFetch;
346
2/4
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 228 times.
 228 if ((write(fd, &cmd, sizeof(Command))) != sizeof(Command)) {
347 ✗ if (not(errno == EAGAIN || errno == EWOULDBLOCK)) {
348 ✗ LogCvmfs(kLogBundleMgr,
349 kLogDebug,
350 "write() on back channel failed unexpectedly");
351 }
352 ✗ return false;
353 } else {
354
1/2
✓ Branch 1 taken 228 times.
✗ Branch 2 not taken.
 228 BlockingSend(fd, path);
355 }
356 228 return true;
357 }
358
359