GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/bundle_mgr.cc
Date: 2026-08-16 02:40:26
Exec Total Coverage
Lines: 204 278 73.4%
Branches: 147 364 40.4%
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 <cstring>
15 #include <memory>
16 #include <string>
17 #include <vector>
18
19 #include "cache.h"
20 #include "catalog_mgr_client.h"
21 #include "fetch.h"
22 #include "file_chunk.h"
23 #include "json_document.h"
24 #include "mountpoint.h"
25 #include "options.h"
26 #include "shortstring.h"
27 #include "util/posix.h"
28
29 namespace {
30 constexpr size_t kDefaultBundlePoolSize = 8;
31
32 // Read the .cvmfsbundle-<basename> file via the cvmfs cache
33 58 BundleFileMgr *LoadBundleFromCvmfs(MountPoint *mp,
34 const PathString &bundle_file_path) {
35
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 catalog::DirectoryEntry dirent;
36
2/4
✓ Branch 2 taken 58 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 58 times.
 58 if (!mp->catalog_mgr()->LookupPath(bundle_file_path, catalog::kLookupDefault,
37 &dirent)) {
38 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: LookupPath failed for %s",
39 ✗ bundle_file_path.ToString().c_str());
40 ✗ return nullptr;
41 }
42 58 cvmfs::Fetcher *fetcher = mp->fetcher();
43
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 58 times.
 58 if (fetcher == nullptr) {
44 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: fetcher is null");
45 ✗ return nullptr;
46 }
47
48 58 CacheManager::Label label;
49
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 label.path = bundle_file_path.ToString();
50
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 label.size = dirent.size();
51 58 label.zip_algorithm = dirent.compression_algorithm();
52
2/4
✓ Branch 2 taken 58 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 58 times.
✗ Branch 6 not taken.
 174 const int fd = fetcher->Fetch(
53
1/2
✓ Branch 2 taken 58 times.
✗ Branch 3 not taken.
 116 CacheManager::LabeledObject(dirent.checksum(), label));
54
1/2
✓ Branch 0 taken 58 times.
✗ Branch 1 not taken.
 58 if (fd < 0) {
55
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: Fetch returned fd=%d", fd);
56 58 return nullptr;
57 }
58
59 ✗ CacheManager *cache_mgr = mp->file_system()->cache_mgr();
60 ✗ std::string content;
61 ✗ content.resize(static_cast<size_t>(dirent.size()));
62 ✗ const int64_t n = cache_mgr->Pread(fd, &content[0], content.size(), 0);
63 ✗ cache_mgr->Close(fd);
64 ✗ if (n < 0 || static_cast<size_t>(n) != content.size()) {
65 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: Pread returned %ld want %zu",
66 static_cast<long>(n), content.size());
67 ✗ return nullptr;
68 }
69
70 // The bundle file may start with a "#%CVMFS_BUNDLE version=..." header
71 // line (per file_bundle.h); strip any leading lines beginning with '#'
72 // before handing off to the strict JSON parser.
73 ✗ size_t json_start = 0;
74 ✗ while (json_start < content.size() && content[json_start] == '#') {
75 ✗ const size_t nl = content.find('\n', json_start);
76 ✗ if (nl == std::string::npos) {
77 ✗ json_start = content.size();
78 ✗ break;
79 }
80 ✗ json_start = nl + 1;
81 }
82 const std::string json_text = (json_start == 0) ? content
83 ✗ : content.substr(json_start);
84
85 ✗ JsonDocument *doc = JsonDocument::Create(json_text);
86 ✗ if (doc == nullptr) {
87 ✗ LogCvmfs(kLogCvmfs, kLogDebug,
88 "BUNDLE-LOAD: JsonDocument::Create failed (size=%zu)",
89 json_text.size());
90 ✗ return nullptr;
91 }
92 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: loaded bundle %s (%zu bytes)",
93 ✗ bundle_file_path.ToString().c_str(), content.size());
94 ✗ return new BundleFileMgr(doc);
95 58 }
96 } // namespace
97
98 261 BundleMgr::BundleMgr(MountPoint *mp)
99 261 : mount_point_(mp)
100 261 , fetcher_threads_()
101 261 , pool_size_(kDefaultBundlePoolSize) {
102 261 atomic_init32(&terminating_);
103 261 pthread_mutex_init(&worker_read_mutex_, nullptr);
104
105 // Pool size override via CVMFS_BUNDLE_POOL_SIZE
106
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 261 if (mount_point_ != nullptr && mount_point_->file_system() != nullptr
107
3/6
✓ Branch 0 taken 261 times.
✗ Branch 1 not taken.
✓ Branch 4 taken 261 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 261 times.
✗ Branch 7 not taken.
 522 && mount_point_->file_system()->options_mgr() != nullptr) {
108 261 std::string opt;
109
3/6
✓ Branch 4 taken 261 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 261 times.
✗ Branch 8 not taken.
✗ Branch 11 not taken.
✓ Branch 12 taken 261 times.
 261 if (mount_point_->file_system()->options_mgr()->GetValue(
110 "CVMFS_BUNDLE_POOL_SIZE", &opt)) {
111 ✗ char *end = nullptr;
112 ✗ const unsigned long n = std::strtoul(opt.c_str(), &end, 10);
113 ✗ if (end != opt.c_str() && n >= 1) {
114 ✗ pool_size_ = static_cast<size_t>(n);
115 }
116 }
117 261 }
118
119 // The queues are created here so that ScheduleTrigger() can already
120 // enqueue before Spawn(); pipe file descriptors, unlike threads, survive
121 // the fuse client's daemonization fork.
122
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 261 MakePipe(pipe_bm_);
123
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 261 MakePipe(pipe_triggers_);
124
125 // Non-blocking writes so TrySendPath/ScheduleTrigger can drop when a
126 // queue is full. Per pipe(7), writes <= PIPE_BUF are atomic on
127 // non-blocking pipes: they either fully succeed or fail with EAGAIN.
128
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 261 int flags = fcntl(pipe_bm_[1], F_GETFL);
129
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 261 fcntl(pipe_bm_[1], F_SETFL, flags | O_NONBLOCK);
130
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 261 flags = fcntl(pipe_triggers_[1], F_GETFL);
131
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 261 fcntl(pipe_triggers_[1], F_SETFL, flags | O_NONBLOCK);
132 261 }
133
134 261 void BundleMgr::Spawn() {
135 261 SpawnFetcherPool();
136
1/2
✓ Branch 0 taken 261 times.
✗ Branch 1 not taken.
 261 if (is_valid_)
137 261 SpawnDispatcher();
138 261 }
139
140 58 bool BundleMgr::ScheduleTrigger(const PathString &path) {
141
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 58 times.
 58 if (not is_valid_) {
142 ✗ LogCvmfs(kLogBundleMgr,
143 kLogDebug,
144 "BundleMgr is not in a valid state. Can't schedule trigger!");
145 ✗ return false;
146 }
147 // A single non-blocking attempt: prefetching is best-effort, so if the
148 // trigger queue is full the request is dropped instead of stalling the
149 // caller (an open() holding the remount fence).
150
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 58 times.
 58 if (not TrySendPath(pipe_triggers_[1], path)) {
151 ✗ LogCvmfs(kLogBundleMgr, kLogDebug, "trigger queue full, dropping %s",
152 ✗ path.ToString().c_str());
153 ✗ return false;
154 }
155 58 return true;
156 }
157
158 /**
159 * Dependency paths in a bundle are absolute from the repository root.
160 * Entries without a leading slash (optionally prefixed with "./") are
161 * resolved relative to the directory holding the bundle file.
162 */
163 174 PathString BundleMgr::NormalizeDependencyPath(const PathString &path,
164 const PathString &parent_path) {
165
2/4
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 174 times.
 174 if (path.StartsWith(PathString("/", 1)))
166 ✗ return path;
167
1/2
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
 174 const PathString relative = path.StartsWith(PathString("./", 2))
168 ? path.Suffix(2)
169
2/6
✗ Branch 0 not taken.
✓ Branch 1 taken 174 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 6 taken 174 times.
✗ Branch 7 not taken.
 174 : path;
170
1/2
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
 174 PathString normalized(parent_path);
171
1/2
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
 174 normalized.Append("/", 1);
172
1/2
✓ Branch 3 taken 174 times.
✗ Branch 4 not taken.
 174 normalized.Append(relative.GetChars(), relative.GetLength());
173
1/2
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
 174 return normalized;
174 174 }
175
176 377 void BundleMgr::JoinFetcherPool() {
177
2/2
✓ Branch 0 taken 116 times.
✓ Branch 1 taken 261 times.
 377 if (pipe_bm_[1] < 0)
178 116 return;
179 // Send one kTerminate per worker. Workers drain all queued kFetch
180 // messages before reaching their kTerminate (FIFO pipe), so we can't
181 // just close the pipe — that would EOF some workers mid-drain.
182
2/2
✓ Branch 1 taken 2088 times.
✓ Branch 2 taken 261 times.
 2349 for (size_t i = 0; i < fetcher_threads_.size(); ++i) {
183 2088 Command cmd = Command::kTerminate;
184 while (true) {
185
1/2
✓ Branch 1 taken 2088 times.
✗ Branch 2 not taken.
 2088 const ssize_t n = ::write(pipe_bm_[1], &cmd, sizeof(Command));
186
1/2
✓ Branch 0 taken 2088 times.
✗ Branch 1 not taken.
 2088 if (n == sizeof(Command))
187 2088 break;
188 ✗ if (errno != EAGAIN && errno != EWOULDBLOCK)
189 ✗ break;
190 }
191 }
192 // Wait for every worker to drain its share of the queue and exit.
193
2/2
✓ Branch 5 taken 2088 times.
✓ Branch 6 taken 261 times.
 2349 for (auto &t : fetcher_threads_) {
194
1/2
✓ Branch 2 taken 2088 times.
✗ Branch 3 not taken.
 2088 pthread_join(*t, nullptr);
195 }
196 261 fetcher_threads_.clear();
197 261 ClosePipe(pipe_bm_);
198 // Mark the pool as gone so that a subsequent call (e.g. the destructor
199 // running after an explicit JoinFetcherPool()) is a no-op instead of a
200 // double close/join.
201 261 pipe_bm_[0] = pipe_bm_[1] = -1;
202 }
203
204 261 void BundleMgr::SpawnFetcherPool() {
205
2/2
✓ Branch 0 taken 2088 times.
✓ Branch 1 taken 261 times.
 2349 for (size_t i = 0; i < pool_size_; ++i) {
206
1/2
✓ Branch 1 taken 2088 times.
✗ Branch 2 not taken.
 2088 std::unique_ptr<pthread_t> thread(new pthread_t());
207 2088 const int res = pthread_create(thread.get(), nullptr, MainBundleMgrFetcher,
208 this);
209
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2088 times.
 2088 if (res != 0) {
210 ✗ LogCvmfs(kLogBundleMgr, kLogDebug,
211 "Thread creation failed! pool_size_=%zu spawned=%zu", pool_size_,
212 i);
213 ✗ is_valid_ = false;
214 ✗ return;
215 }
216
1/2
✓ Branch 2 taken 2088 times.
✗ Branch 3 not taken.
 2088 fetcher_threads_.emplace_back(std::move(thread));
217
1/2
✓ Branch 1 taken 2088 times.
✗ Branch 2 not taken.
 2088 }
218 }
219
220 261 void BundleMgr::SpawnDispatcher() {
221 261 dispatcher_thread_.reset(new pthread_t());
222 261 const int res = pthread_create(dispatcher_thread_.get(), nullptr,
223 MainBundleMgrDispatcher, this);
224
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 261 times.
 261 if (res != 0) {
225 ✗ LogCvmfs(kLogBundleMgr, kLogDebug, "Dispatcher thread creation failed!");
226 ✗ dispatcher_thread_.reset();
227 ✗ is_valid_ = false;
228 }
229 261 }
230
231 319 void BundleMgr::JoinDispatcher() {
232
2/2
✓ Branch 0 taken 58 times.
✓ Branch 1 taken 261 times.
 319 if (pipe_triggers_[1] < 0)
233 58 return;
234 // The dispatcher only exists once Spawn() has run; without it there is
235 // just the pipe to close.
236
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 261 if (dispatcher_thread_) {
237 261 Command cmd = Command::kTerminate;
238 while (true) {
239
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 261 const ssize_t n = ::write(pipe_triggers_[1], &cmd, sizeof(Command));
240
1/2
✓ Branch 0 taken 261 times.
✗ Branch 1 not taken.
 261 if (n == sizeof(Command))
241 261 break;
242 ✗ if (errno != EAGAIN && errno != EWOULDBLOCK)
243 ✗ break;
244 }
245
1/2
✓ Branch 2 taken 261 times.
✗ Branch 3 not taken.
 261 pthread_join(*dispatcher_thread_, nullptr);
246 261 dispatcher_thread_.reset();
247 }
248 261 ClosePipe(pipe_triggers_);
249 261 pipe_triggers_[0] = pipe_triggers_[1] = -1;
250 }
251
252 /**
253 * Loads the bundle spec that belongs to the given trigger file and enqueues
254 * its dependencies for the fetcher pool. Runs on the dispatcher thread.
255 */
256 58 void BundleMgr::ProcessTrigger(const PathString &trigger_path) {
257
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 const NameString fname = GetFileName(trigger_path);
258
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 const PathString parent_path = GetParentPath(trigger_path);
259 // There is a naming convention regarding the name of the file with the
260 // contents of the bundle
261
2/4
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 58 times.
✗ Branch 5 not taken.
 116 const PathString bundle_file_path(parent_path.ToString() + "/.cvmfsbundle-"
262
3/6
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 58 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 58 times.
✗ Branch 8 not taken.
 174 + fname.ToString());
263
264 const std::unique_ptr<BundleFileMgr> bfm(
265
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 LoadBundleFromCvmfs(mount_point_, bundle_file_path));
266
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 if (bfm == nullptr) {
267
1/2
✓ Branch 2 taken 58 times.
✗ Branch 3 not taken.
 58 LogCvmfs(kLogCvmfs, kLogDebug, "Couldn't fetch bundle associated to %s",
268
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 116 trigger_path.ToString().c_str());
269 58 return;
270 }
271 ✗ EnqueueDependencies(bfm.get(), parent_path);
272
4/8
✗ Branch 1 not taken.
✓ Branch 2 taken 58 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 58 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 58 times.
✗ Branch 10 not taken.
✓ Branch 11 taken 58 times.
 232 }
273
274 232 void BundleMgr::EnqueueDependencies(BundleFileMgr *bfm,
275 const PathString &parent_path) {
276
4/6
✓ Branch 1 taken 232 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 232 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 58 times.
✓ Branch 7 taken 174 times.
 232 while (auto file = bfm->GetNext()) {
277
1/2
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
 174 const PathString path = NormalizeDependencyPath(file, parent_path);
278 // A single non-blocking attempt: if the dependency queue is full the
279 // entry is dropped (prefetching is best-effort) instead of spinning.
280
2/4
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 174 times.
 174 if (not TrySendPath(pipe_bm_[1], path)) {
281 ✗ LogCvmfs(kLogBundleMgr, kLogDebug, "dependency queue full, dropping %s",
282 ✗ path.ToString().c_str());
283 }
284
2/2
✓ Branch 2 taken 174 times.
✓ Branch 3 taken 58 times.
 406 }
285 58 }
286
287 174 void BundleMgr::FetchPath(const PathString &path) {
288
1/2
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
 174 catalog::DirectoryEntry dirent;
289
1/2
✓ Branch 2 taken 174 times.
✗ Branch 3 not taken.
 174 const bool found = mount_point_->catalog_mgr()->LookupPath(
290 path, catalog::kLookupDefault, &dirent);
291 174 cvmfs::Fetcher *this_fetcher = dirent.IsExternalFile()
292
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 174 times.
 174 ? mount_point_->external_fetcher()
293 174 : mount_point_->fetcher();
294
2/4
✓ Branch 0 taken 174 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 174 times.
 174 if (not(found and this_fetcher)) {
295 ✗ LogCvmfs(kLogCvmfs, kLogDebug,
296 "BUNDLE-FETCH: lookup failed for %s (found=%d)",
297 ✗ path.ToString().c_str(), int(found));
298 ✗ return;
299 }
300
1/2
✓ Branch 2 taken 174 times.
✗ Branch 3 not taken.
 174 LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-FETCH: prefetching %s",
301
1/2
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
 348 path.ToString().c_str());
302
303
2/2
✓ Branch 1 taken 87 times.
✓ Branch 2 taken 87 times.
 174 if (dirent.IsChunkedFile()) {
304 // Files above the chunking threshold are stored as per-chunk objects;
305 // their bulk object only exists if the repository sets
306 // CVMFS_GENERATE_LEGACY_BULK_CHUNKS. Fetch the chunks, exactly like the
307 // read path does.
308
1/2
✓ Branch 1 taken 87 times.
✗ Branch 2 not taken.
 87 FileChunkList chunks;
309
1/2
✓ Branch 3 taken 87 times.
✗ Branch 4 not taken.
 87 if (!mount_point_->catalog_mgr()->ListFileChunks(
310 path, dirent.hash_algorithm(), &chunks)
311
3/6
✓ Branch 0 taken 87 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 58 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 58 times.
 87 || chunks.IsEmpty()) {
312 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-FETCH: no chunks found for %s",
313 ✗ path.ToString().c_str());
314 ✗ return;
315 }
316
2/2
✓ Branch 1 taken 232 times.
✓ Branch 2 taken 87 times.
 319 for (unsigned i = 0; i < chunks.size(); ++i) {
317 232 CacheManager::Label label;
318
1/2
✓ Branch 1 taken 261 times.
✗ Branch 2 not taken.
 232 label.path = path.ToString();
319 261 label.size = chunks.AtPtr(i)->size();
320 261 label.zip_algorithm = dirent.compression_algorithm();
321 261 label.flags |= CacheManager::kLabelChunked;
322
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 261 times.
 261 if (mount_point_->catalog_mgr()->volatile_flag())
323 ✗ label.flags |= CacheManager::kLabelVolatile;
324
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 261 times.
 261 if (dirent.IsExternalFile()) {
325 ✗ label.flags |= CacheManager::kLabelExternal;
326 ✗ label.range_offset = chunks.AtPtr(i)->offset();
327 }
328
2/4
✓ Branch 2 taken 261 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 261 times.
✗ Branch 6 not taken.
 783 const int fd = this_fetcher->Fetch(
329
1/2
✓ Branch 3 taken 261 times.
✗ Branch 4 not taken.
 522 CacheManager::LabeledObject(chunks.AtPtr(i)->content_hash(), label));
330
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 261 times.
 261 if (fd >= 0)
331 ✗ mount_point_->file_system()->cache_mgr()->Close(fd);
332 261 }
333 87 return;
334 87 }
335
336 87 CacheManager::Label label;
337
1/2
✓ Branch 1 taken 87 times.
✗ Branch 2 not taken.
 87 label.path = path.ToString();
338
1/2
✓ Branch 1 taken 87 times.
✗ Branch 2 not taken.
 87 label.size = dirent.size();
339 87 label.zip_algorithm = dirent.compression_algorithm();
340
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 87 times.
 87 if (mount_point_->catalog_mgr()->volatile_flag())
341 ✗ label.flags |= CacheManager::kLabelVolatile;
342
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 87 times.
 87 if (dirent.IsExternalFile())
343 ✗ label.flags |= CacheManager::kLabelExternal;
344
2/4
✓ Branch 2 taken 87 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 87 times.
✗ Branch 6 not taken.
 261 const int fd = this_fetcher->Fetch(
345
1/2
✓ Branch 2 taken 87 times.
✗ Branch 3 not taken.
 174 CacheManager::LabeledObject(dirent.checksum(), label));
346
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 87 times.
 87 if (fd >= 0)
347 ✗ mount_point_->file_system()->cache_mgr()->Close(fd);
348
2/2
✓ Branch 2 taken 87 times.
✓ Branch 3 taken 87 times.
 174 }
349
350 261 void *BundleMgr::MainBundleMgrDispatcher(void *data) {
351 #ifndef __APPLE__
352 261 pthread_setname_np(pthread_self(), "bm_dispatch");
353 #endif
354 261 BundleMgr *mgr = static_cast<BundleMgr *>(data);
355 261 const int rfd = mgr->pipe_triggers_[0];
356
357 // Single reader on this pipe, so no receive mutex is needed here
358 while (true) {
359 319 Command cmd = Command::kTerminate;
360
1/2
✓ Branch 1 taken 319 times.
✗ Branch 2 not taken.
 319 const ssize_t n = read(rfd, &cmd, sizeof(Command));
361
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 319 times.
 319 if (n != static_cast<ssize_t>(sizeof(Command)))
362 ✗ break;
363
2/2
✓ Branch 0 taken 261 times.
✓ Branch 1 taken 58 times.
 319 if (cmd != Command::kFetch)
364 261 break;
365
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 const PathString path = mgr->ReceivePath(rfd);
366 // While terminating, drain the queue without processing so that
367 // unmounting does not wait for spec downloads
368
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 if (atomic_read32(&mgr->terminating_) == 0)
369
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 mgr->ProcessTrigger(path);
370 58 }
371
372 261 pthread_exit(nullptr);
373 }
374
375 2088 void *BundleMgr::MainBundleMgrFetcher(void *data) {
376 #ifndef __APPLE__
377 2088 pthread_setname_np(pthread_self(), "bm_fetcher");
378 #endif
379 2088 BundleMgr *mgr = static_cast<BundleMgr *>(data);
380 2088 const int rfd = mgr->pipe_bm_[0];
381
382 while (true) {
383 2262 Command cmd = Command::kTerminate;
384 2262 PathString path;
385 2262 bool got_path = false;
386 2262 bool eof = false;
387
388 // Atomically receive cmd + (optional) path payload. The whole receipt
389 // is under worker_read_mutex_ so messages aren't interleaved between
390 // workers reading from the shared pipe.
391 2262 pthread_mutex_lock(&mgr->worker_read_mutex_);
392
1/2
✓ Branch 1 taken 2262 times.
✗ Branch 2 not taken.
 2262 const ssize_t n = read(rfd, &cmd, sizeof(Command));
393
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2262 times.
 2262 if (n != static_cast<ssize_t>(sizeof(Command))) {
394 ✗ eof = true;
395
2/2
✓ Branch 0 taken 174 times.
✓ Branch 1 taken 2088 times.
 2262 } else if (cmd == Command::kFetch) {
396
2/4
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 174 times.
✗ Branch 5 not taken.
 174 path = mgr->ReceivePath(rfd);
397 174 got_path = true;
398 }
399 2262 pthread_mutex_unlock(&mgr->worker_read_mutex_);
400
401
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2262 times.
 2262 if (eof)
402 ✗ break;
403
404 2262 bool terminate = false;
405
2/2
✓ Branch 0 taken 174 times.
✓ Branch 1 taken 2088 times.
 2262 switch (cmd) {
406 174 case Command::kFetch: {
407
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 174 times.
 174 if (!got_path) {
408 ✗ terminate = true;
409 ✗ break;
410 }
411 // While terminating, drain the queue without fetching so that
412 // unmounting does not wait for pending downloads
413
1/2
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
 174 if (atomic_read32(&mgr->terminating_) == 0)
414
1/2
✓ Branch 1 taken 174 times.
✗ Branch 2 not taken.
 174 mgr->FetchPath(path);
415 174 } break;
416 2088 case Command::kTerminate:
417 default:
418 2088 terminate = true;
419 2088 break;
420 }
421
2/2
✓ Branch 0 taken 2088 times.
✓ Branch 1 taken 174 times.
 2262 if (terminate) {
422 2088 break;
423 }
424
2/2
✓ Branch 1 taken 174 times.
✓ Branch 2 taken 2088 times.
 2436 }
425
426 2088 pthread_exit(nullptr);
427 }
428
429 290 PathString BundleMgr::ReceivePath(int fd) const {
430
1/2
✓ Branch 1 taken 290 times.
✗ Branch 2 not taken.
 290 const std::string buffer = BlockingReceive(fd);
431
1/2
✓ Branch 1 taken 290 times.
✗ Branch 2 not taken.
 290 assert(buffer.size() > 0 && "A path can't be empty");
432
1/2
✓ Branch 1 taken 290 times.
✗ Branch 2 not taken.
 580 return PathString(buffer);
433 290 }
434
435 48981 bool BundleMgr::TrySendPath(int fd, const PathString &path) const {
436 // The whole message (command + length + payload) is sent as a single
437 // write: per pipe(7), writes <= PIPE_BUF to a non-blocking pipe are
438 // atomic, they either fully succeed or fail with EAGAIN. Sending the
439 // parts separately could hit a full queue in the middle of a message
440 // and corrupt the stream for all readers.
441 48981 const Command cmd = Command::kFetch;
442 48981 const size_t length = path.GetLength();
443 48981 const size_t msg_size = sizeof(cmd) + sizeof(length) + length;
444
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 48981 times.
 48981 if (msg_size > PIPE_BUF) {
445 ✗ LogCvmfs(kLogBundleMgr, kLogDebug,
446 "path too long for the work queue, dropping %s",
447 ✗ path.ToString().c_str());
448 ✗ return false;
449 }
450 char msg[PIPE_BUF];
451 48981 memcpy(msg, &cmd, sizeof(cmd));
452 48981 memcpy(msg + sizeof(cmd), &length, sizeof(length));
453 48981 memcpy(msg + sizeof(cmd) + sizeof(length), path.GetChars(), length);
454
455
1/2
✓ Branch 1 taken 48981 times.
✗ Branch 2 not taken.
 48981 const ssize_t n = write(fd, msg, msg_size);
456
2/2
✓ Branch 0 taken 48952 times.
✓ Branch 1 taken 29 times.
 48981 if (n == static_cast<ssize_t>(msg_size))
457 48952 return true;
458
2/6
✓ Branch 0 taken 29 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 29 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 29 if ((n < 0) && not(errno == EAGAIN || errno == EWOULDBLOCK)) {
459 ✗ LogCvmfs(kLogBundleMgr, kLogDebug,
460 "write() on the work queue failed unexpectedly (errno=%d)",
461 ✗ errno);
462 }
463 29 return false;
464 }
465