GCC Code Coverage Report

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