GCC Code Coverage Report

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