GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/bundle_mgr.cc
Date: 2026-11-08 02:41:18
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 34 BundleFileMgr *LoadBundleFromCvmfs(MountPoint *mp,
34 const PathString &bundle_file_path) {
35
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 catalog::DirectoryEntry dirent;
36
2/4
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 34 times.
 34 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 34 cvmfs::Fetcher *fetcher = mp->fetcher();
43
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 34 times.
 34 if (fetcher == nullptr) {
44 ✗ LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: fetcher is null");
45 ✗ return nullptr;
46 }
47
48 34 CacheManager::Label label;
49
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 label.path = bundle_file_path.ToString();
50
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 label.size = dirent.size();
51 34 label.zip_algorithm = dirent.compression_algorithm();
52
2/4
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 34 times.
✗ Branch 6 not taken.
 102 const int fd = fetcher->Fetch(
53
1/2
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
 68 CacheManager::LabeledObject(dirent.checksum(), label));
54
1/2
✓ Branch 0 taken 34 times.
✗ Branch 1 not taken.
 34 if (fd < 0) {
55
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-LOAD: Fetch returned fd=%d", fd);
56 34 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 34 }
96 } // namespace
97
98 153 BundleMgr::BundleMgr(MountPoint *mp)
99 153 : mount_point_(mp), fetcher_threads_(), pool_size_(kDefaultBundlePoolSize) {
100 153 terminating_.store(0);
101 153 pthread_mutex_init(&worker_read_mutex_, nullptr);
102
103 // Pool size override via CVMFS_BUNDLE_POOL_SIZE
104
1/2
✓ Branch 1 taken 153 times.
✗ Branch 2 not taken.
 153 if (mount_point_ != nullptr && mount_point_->file_system() != nullptr
105
3/6
✓ Branch 0 taken 153 times.
✗ Branch 1 not taken.
✓ Branch 4 taken 153 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 153 times.
✗ Branch 7 not taken.
 306 && mount_point_->file_system()->options_mgr() != nullptr) {
106 153 std::string opt;
107
3/6
✓ Branch 4 taken 153 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 153 times.
✗ Branch 8 not taken.
✗ Branch 11 not taken.
✓ Branch 12 taken 153 times.
 153 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 153 }
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 153 times.
✗ Branch 2 not taken.
 153 MakePipe(pipe_bm_);
121
1/2
✓ Branch 1 taken 153 times.
✗ Branch 2 not taken.
 153 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 153 times.
✗ Branch 2 not taken.
 153 int flags = fcntl(pipe_bm_[1], F_GETFL);
127
1/2
✓ Branch 1 taken 153 times.
✗ Branch 2 not taken.
 153 fcntl(pipe_bm_[1], F_SETFL, flags | O_NONBLOCK);
128
1/2
✓ Branch 1 taken 153 times.
✗ Branch 2 not taken.
 153 flags = fcntl(pipe_triggers_[1], F_GETFL);
129
1/2
✓ Branch 1 taken 153 times.
✗ Branch 2 not taken.
 153 fcntl(pipe_triggers_[1], F_SETFL, flags | O_NONBLOCK);
130 153 }
131
132 153 void BundleMgr::Spawn() {
133 153 SpawnFetcherPool();
134
1/2
✓ Branch 0 taken 153 times.
✗ Branch 1 not taken.
 153 if (is_valid_)
135 153 SpawnDispatcher();
136 153 }
137
138 34 bool BundleMgr::ScheduleTrigger(const PathString &path) {
139
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 34 times.
 34 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 34 times.
 34 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 34 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 102 PathString BundleMgr::NormalizeDependencyPath(const PathString &path,
162 const PathString &parent_path) {
163
2/4
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 102 times.
 102 if (path.StartsWith(PathString("/", 1)))
164 ✗ return path;
165
1/2
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
 102 const PathString relative = path.StartsWith(PathString("./", 2))
166 ? path.Suffix(2)
167
2/6
✗ Branch 0 not taken.
✓ Branch 1 taken 102 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 6 taken 102 times.
✗ Branch 7 not taken.
 102 : path;
168
1/2
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
 102 PathString normalized(parent_path);
169
1/2
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
 102 normalized.Append("/", 1);
170
1/2
✓ Branch 3 taken 102 times.
✗ Branch 4 not taken.
 102 normalized.Append(relative.GetChars(), relative.GetLength());
171
1/2
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
 102 return normalized;
172 102 }
173
174 221 void BundleMgr::JoinFetcherPool() {
175
2/2
✓ Branch 0 taken 68 times.
✓ Branch 1 taken 153 times.
 221 if (pipe_bm_[1] < 0)
176 68 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 1224 times.
✓ Branch 2 taken 153 times.
 1377 for (size_t i = 0; i < fetcher_threads_.size(); ++i) {
181 1224 Command cmd = Command::kTerminate;
182 while (true) {
183
1/2
✓ Branch 1 taken 1224 times.
✗ Branch 2 not taken.
 1224 const ssize_t n = ::write(pipe_bm_[1], &cmd, sizeof(Command));
184
1/2
✓ Branch 0 taken 1224 times.
✗ Branch 1 not taken.
 1224 if (n == sizeof(Command))
185 1224 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 1224 times.
✓ Branch 6 taken 153 times.
 1377 for (auto &t : fetcher_threads_) {
192
1/2
✓ Branch 2 taken 1224 times.
✗ Branch 3 not taken.
 1224 pthread_join(*t, nullptr);
193 }
194 153 fetcher_threads_.clear();
195 153 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 153 pipe_bm_[0] = pipe_bm_[1] = -1;
200 }
201
202 153 void BundleMgr::SpawnFetcherPool() {
203
2/2
✓ Branch 0 taken 1224 times.
✓ Branch 1 taken 153 times.
 1377 for (size_t i = 0; i < pool_size_; ++i) {
204
1/2
✓ Branch 1 taken 1224 times.
✗ Branch 2 not taken.
 1224 std::unique_ptr<pthread_t> thread(new pthread_t());
205 1224 const int res = pthread_create(thread.get(), nullptr, MainBundleMgrFetcher,
206 this);
207
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1224 times.
 1224 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 1224 times.
✗ Branch 3 not taken.
 1224 fetcher_threads_.emplace_back(std::move(thread));
215
1/2
✓ Branch 1 taken 1224 times.
✗ Branch 2 not taken.
 1224 }
216 }
217
218 153 void BundleMgr::SpawnDispatcher() {
219 153 dispatcher_thread_.reset(new pthread_t());
220 153 const int res = pthread_create(dispatcher_thread_.get(), nullptr,
221 MainBundleMgrDispatcher, this);
222
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 153 times.
 153 if (res != 0) {
223 ✗ LogCvmfs(kLogBundleMgr, kLogDebug, "Dispatcher thread creation failed!");
224 ✗ dispatcher_thread_.reset();
225 ✗ is_valid_ = false;
226 }
227 153 }
228
229 187 void BundleMgr::JoinDispatcher() {
230
2/2
✓ Branch 0 taken 34 times.
✓ Branch 1 taken 153 times.
 187 if (pipe_triggers_[1] < 0)
231 34 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 153 times.
✗ Branch 2 not taken.
 153 if (dispatcher_thread_) {
235 153 Command cmd = Command::kTerminate;
236 while (true) {
237
1/2
✓ Branch 1 taken 153 times.
✗ Branch 2 not taken.
 153 const ssize_t n = ::write(pipe_triggers_[1], &cmd, sizeof(Command));
238
1/2
✓ Branch 0 taken 153 times.
✗ Branch 1 not taken.
 153 if (n == sizeof(Command))
239 153 break;
240 ✗ if (errno != EAGAIN && errno != EWOULDBLOCK)
241 ✗ break;
242 }
243
1/2
✓ Branch 2 taken 153 times.
✗ Branch 3 not taken.
 153 pthread_join(*dispatcher_thread_, nullptr);
244 153 dispatcher_thread_.reset();
245 }
246 153 ClosePipe(pipe_triggers_);
247 153 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 34 void BundleMgr::ProcessTrigger(const PathString &trigger_path) {
255
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 const NameString fname = GetFileName(trigger_path);
256
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 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 34 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 34 times.
✗ Branch 5 not taken.
 68 const PathString bundle_file_path(parent_path.ToString() + "/.cvmfsbundle-"
260
3/6
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 34 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 34 times.
✗ Branch 8 not taken.
 102 + fname.ToString());
261
262 const std::unique_ptr<BundleFileMgr> bfm(
263
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 LoadBundleFromCvmfs(mount_point_, bundle_file_path));
264
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 if (bfm == nullptr) {
265
1/2
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
 34 LogCvmfs(kLogCvmfs, kLogDebug, "Couldn't fetch bundle associated to %s",
266
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 68 trigger_path.ToString().c_str());
267 34 return;
268 }
269 ✗ EnqueueDependencies(bfm.get(), parent_path);
270
4/8
✗ Branch 1 not taken.
✓ Branch 2 taken 34 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 34 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 34 times.
✗ Branch 10 not taken.
✓ Branch 11 taken 34 times.
 136 }
271
272 136 void BundleMgr::EnqueueDependencies(BundleFileMgr *bfm,
273 const PathString &parent_path) {
274
4/6
✓ Branch 1 taken 136 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 136 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 34 times.
✓ Branch 7 taken 102 times.
 136 while (auto file = bfm->GetNext()) {
275
1/2
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
 102 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 102 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 102 times.
 102 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 102 times.
✓ Branch 3 taken 34 times.
 238 }
283 34 }
284
285 102 void BundleMgr::FetchPath(const PathString &path) {
286
1/2
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
 102 catalog::DirectoryEntry dirent;
287
1/2
✓ Branch 2 taken 102 times.
✗ Branch 3 not taken.
 102 const bool found = mount_point_->catalog_mgr()->LookupPath(
288 path, catalog::kLookupDefault, &dirent);
289 102 cvmfs::Fetcher *this_fetcher = dirent.IsExternalFile()
290
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 102 times.
 102 ? mount_point_->external_fetcher()
291 102 : mount_point_->fetcher();
292
2/4
✓ Branch 0 taken 102 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 102 times.
 102 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 102 times.
✗ Branch 3 not taken.
 102 LogCvmfs(kLogCvmfs, kLogDebug, "BUNDLE-FETCH: prefetching %s",
299
1/2
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
 204 path.ToString().c_str());
300
301
2/2
✓ Branch 1 taken 51 times.
✓ Branch 2 taken 51 times.
 102 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 51 times.
✗ Branch 2 not taken.
 51 FileChunkList chunks;
307
1/2
✓ Branch 3 taken 51 times.
✗ Branch 4 not taken.
 51 if (!mount_point_->catalog_mgr()->ListFileChunks(
308 path, dirent.hash_algorithm(), &chunks)
309
3/6
✓ Branch 0 taken 51 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 51 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 51 times.
 51 || 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 153 times.
✓ Branch 2 taken 51 times.
 204 for (unsigned i = 0; i < chunks.size(); ++i) {
315 153 CacheManager::Label label;
316
1/2
✓ Branch 1 taken 153 times.
✗ Branch 2 not taken.
 153 label.path = path.ToString();
317 153 label.size = chunks.AtPtr(i)->size();
318 153 label.zip_algorithm = dirent.compression_algorithm();
319 153 label.flags |= CacheManager::kLabelChunked;
320
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 153 times.
 153 if (mount_point_->catalog_mgr()->volatile_flag())
321 ✗ label.flags |= CacheManager::kLabelVolatile;
322
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 153 times.
 153 if (dirent.IsExternalFile()) {
323 ✗ label.flags |= CacheManager::kLabelExternal;
324 ✗ label.range_offset = chunks.AtPtr(i)->offset();
325 }
326
2/4
✓ Branch 2 taken 153 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 153 times.
✗ Branch 6 not taken.
 459 const int fd = this_fetcher->Fetch(
327
1/2
✓ Branch 3 taken 153 times.
✗ Branch 4 not taken.
 306 CacheManager::LabeledObject(chunks.AtPtr(i)->content_hash(), label));
328
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 153 times.
 153 if (fd >= 0)
329 ✗ mount_point_->file_system()->cache_mgr()->Close(fd);
330 153 }
331 51 return;
332 51 }
333
334 51 CacheManager::Label label;
335
1/2
✓ Branch 1 taken 51 times.
✗ Branch 2 not taken.
 51 label.path = path.ToString();
336
1/2
✓ Branch 1 taken 51 times.
✗ Branch 2 not taken.
 51 label.size = dirent.size();
337 51 label.zip_algorithm = dirent.compression_algorithm();
338
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 51 times.
 51 if (mount_point_->catalog_mgr()->volatile_flag())
339 ✗ label.flags |= CacheManager::kLabelVolatile;
340
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 51 times.
 51 if (dirent.IsExternalFile())
341 ✗ label.flags |= CacheManager::kLabelExternal;
342
2/4
✓ Branch 2 taken 51 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 51 times.
✗ Branch 6 not taken.
 153 const int fd = this_fetcher->Fetch(
343
1/2
✓ Branch 2 taken 51 times.
✗ Branch 3 not taken.
 102 CacheManager::LabeledObject(dirent.checksum(), label));
344
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 51 times.
 51 if (fd >= 0)
345 ✗ mount_point_->file_system()->cache_mgr()->Close(fd);
346
2/2
✓ Branch 2 taken 51 times.
✓ Branch 3 taken 51 times.
 102 }
347
348 153 void *BundleMgr::MainBundleMgrDispatcher(void *data) {
349 #ifndef __APPLE__
350 153 pthread_setname_np(pthread_self(), "bm_dispatch");
351 #endif
352 153 BundleMgr *mgr = static_cast<BundleMgr *>(data);
353 153 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 187 Command cmd = Command::kTerminate;
358
1/2
✓ Branch 1 taken 187 times.
✗ Branch 2 not taken.
 187 const ssize_t n = read(rfd, &cmd, sizeof(Command));
359
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 187 times.
 187 if (n != static_cast<ssize_t>(sizeof(Command)))
360 ✗ break;
361
2/2
✓ Branch 0 taken 153 times.
✓ Branch 1 taken 34 times.
 187 if (cmd != Command::kFetch)
362 153 break;
363
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 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 34 times.
✗ Branch 2 not taken.
 68 if (mgr->terminating_.load() == 0)
367
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 mgr->ProcessTrigger(path);
368 34 }
369
370 153 pthread_exit(nullptr);
371 }
372
373 1224 void *BundleMgr::MainBundleMgrFetcher(void *data) {
374 #ifndef __APPLE__
375 1224 pthread_setname_np(pthread_self(), "bm_fetcher");
376 #endif
377 1224 BundleMgr *mgr = static_cast<BundleMgr *>(data);
378 1224 const int rfd = mgr->pipe_bm_[0];
379
380 while (true) {
381 1326 Command cmd = Command::kTerminate;
382 1326 PathString path;
383 1326 bool got_path = false;
384 1326 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 1326 pthread_mutex_lock(&mgr->worker_read_mutex_);
390
1/2
✓ Branch 1 taken 1326 times.
✗ Branch 2 not taken.
 1326 const ssize_t n = read(rfd, &cmd, sizeof(Command));
391
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1326 times.
 1326 if (n != static_cast<ssize_t>(sizeof(Command))) {
392 ✗ eof = true;
393
2/2
✓ Branch 0 taken 102 times.
✓ Branch 1 taken 1224 times.
 1326 } else if (cmd == Command::kFetch) {
394
2/4
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 102 times.
✗ Branch 5 not taken.
 102 path = mgr->ReceivePath(rfd);
395 102 got_path = true;
396 }
397 1326 pthread_mutex_unlock(&mgr->worker_read_mutex_);
398
399
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1309 times.
 1309 if (eof)
400 ✗ break;
401
402 1309 bool terminate = false;
403
2/2
✓ Branch 0 taken 102 times.
✓ Branch 1 taken 1207 times.
 1309 switch (cmd) {
404 102 case Command::kFetch: {
405
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 102 times.
 102 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 102 times.
✗ Branch 2 not taken.
 204 if (mgr->terminating_.load() == 0)
412
1/2
✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
 102 mgr->FetchPath(path);
413 102 } break;
414 1207 case Command::kTerminate:
415 default:
416 1207 terminate = true;
417 1207 break;
418 }
419
2/2
✓ Branch 0 taken 1224 times.
✓ Branch 1 taken 85 times.
 1309 if (terminate) {
420 1224 break;
421 }
422
2/2
✓ Branch 1 taken 102 times.
✓ Branch 2 taken 1224 times.
 1411 }
423
424 1224 pthread_exit(nullptr);
425 }
426
427 170 PathString BundleMgr::ReceivePath(int fd) const {
428
1/2
✓ Branch 1 taken 170 times.
✗ Branch 2 not taken.
 170 const std::string buffer = BlockingReceive(fd);
429
1/2
✓ Branch 1 taken 170 times.
✗ Branch 2 not taken.
 170 assert(buffer.size() > 0 && "A path can't be empty");
430
1/2
✓ Branch 1 taken 170 times.
✗ Branch 2 not taken.
 340 return PathString(buffer);
431 170 }
432
433 28713 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 28713 const Command cmd = Command::kFetch;
440 28713 const size_t length = path.GetLength();
441 28713 const size_t msg_size = sizeof(cmd) + sizeof(length) + length;
442
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 28713 times.
 28713 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 28713 memcpy(msg, &cmd, sizeof(cmd));
450 28713 memcpy(msg + sizeof(cmd), &length, sizeof(length));
451 28713 memcpy(msg + sizeof(cmd) + sizeof(length), path.GetChars(), length);
452
453
1/2
✓ Branch 1 taken 28713 times.
✗ Branch 2 not taken.
 28713 const ssize_t n = write(fd, msg, msg_size);
454
2/2
✓ Branch 0 taken 28696 times.
✓ Branch 1 taken 17 times.
 28713 if (n == static_cast<ssize_t>(msg_size))
455 28696 return true;
456
2/6
✓ Branch 0 taken 17 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 17 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 17 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 17 return false;
461 }
462