GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/bundle_mgr.h
Date: 2026-08-16 02:40:26
Exec Total Coverage
Lines: 29 29 100.0%
Branches: 8 16 50.0%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #ifndef CVMFS_BUNDLE_MGR_H_
6 #define CVMFS_BUNDLE_MGR_H_
7
8 #include <limits.h>
9 #include <pthread.h>
10
11 #include <cstddef>
12 #include <memory>
13 #include <type_traits>
14 #include <vector>
15
16 #include "duplex_testing.h"
17 #include "file_bundle.h"
18 #include "mountpoint.h"
19 #include "shortstring.h"
20 #include "util/atomic.h"
21 #include "util/posix.h"
22 #include "util/single_copy.h"
23
24 class MockFetcher;
25
26 /**
27 * Long-lived, best-effort prefetcher for file bundles. One instance per
28 * mount point (created when CVMFS_PREFETCH_FILEBUNDLES is on) owns a
29 * dispatcher thread and a pool of fetcher threads for the lifetime of the
30 * mount. open() calls of a bundle trigger merely enqueue the trigger path
31 * via ScheduleTrigger() and return; spec loading and dependency downloads
32 * happen entirely on the background threads.
33 *
34 * The constructor only sets up the bounded queues; the threads are started
35 * by Spawn(). The fuse client daemonizes between initialization and
36 * cvmfs::Spawn(), and threads do not survive the fork, so Spawn() must not
37 * be called before then. libcvmfs never forks and calls Spawn() right after
38 * creating the mount point.
39 */
40 class BundleMgr : SingleCopy {
41 friend class T_BundleMgr;
42 FRIEND_TEST(T_BundleMgr, ExchangeCT);
43 FRIEND_TEST(T_BundleMgr, ExchangePathString);
44 FRIEND_TEST(T_BundleMgr, ReceivePathLongerThanPipeBuf);
45 FRIEND_TEST(T_BundleMgr, EnqueueDependencies);
46 FRIEND_TEST(T_BundleMgr, FetchChunked);
47 FRIEND_TEST(T_BundleMgr, ScheduleTrigger);
48 FRIEND_TEST(T_BundleMgr, ScheduleTriggerBeforeSpawn);
49 FRIEND_TEST(T_BundleMgr, TrySendPathDropsWhenFull);
50
51 public:
52 explicit BundleMgr(MountPoint *mp);
53 986 virtual ~BundleMgr() {
54 522 atomic_write32(&terminating_, 1);
55 522 JoinDispatcher();
56 522 JoinFetcherPool();
57 522 pthread_mutex_destroy(&worker_read_mutex_);
58 986 }
59
60 /**
61 * Starts the dispatcher thread and the fetcher pool. Triggers scheduled
62 * before Spawn() wait in the queue until the threads come up.
63 */
64 void Spawn();
65
66 /**
67 * Hands a trigger file over to the background prefetcher. Never blocks:
68 * if the trigger queue is full the request is dropped (prefetching is
69 * best-effort). Returns whether the trigger was enqueued.
70 */
71 bool ScheduleTrigger(const PathString &path);
72 464 explicit operator bool() const { return is_valid_; }
73
74 private:
75 static void *MainBundleMgrFetcher(void *data);
76 static void *MainBundleMgrDispatcher(void *data);
77 void SpawnFetcherPool();
78 void JoinFetcherPool();
79 void SpawnDispatcher();
80 void JoinDispatcher();
81 void ProcessTrigger(const PathString &trigger_path);
82 void EnqueueDependencies(BundleFileMgr *bfm, const PathString &parent_path);
83 PathString ReceivePath(int fd) const;
84 bool TrySendPath(int fd, const PathString &path) const;
85 static PathString NormalizeDependencyPath(const PathString &path,
86 const PathString &parent_path);
87
88 void FetchPath(const PathString &path);
89
90 // CT stands for contiguous type
91 template<typename CT,
92 typename = typename std::enable_if<
93 std::is_trivially_copyable<CT>::value>::type>
94 232 void BlockingSend(int fd, const CT &obj, size_t size = sizeof(CT)) const {
95 typedef typename std::remove_cv<CT>::type T;
96 static_assert(
97 std::is_trivially_copyable<T>::value,
98 "Can't directly send non trivially copyable types over a pipe");
99 static_assert(sizeof(T) == sizeof(CT), "CT illformed");
100 static_assert(
101 sizeof(T) <= PIPE_BUF,
102 "Type too big to be guaranteed atomic transmission over a pipe");
103
104 232 const T *ptr = reinterpret_cast<const T *>(&obj);
105 232 WritePipe(fd, ptr, size);
106 232 }
107
108 58 void BlockingSend(int fd, const PathString &path) const {
109 58 const size_t size = path.GetLength();
110
1/2
✓ Branch 1 taken 58 times.
✗ Branch 2 not taken.
 58 BlockingSend(fd, size);
111
1/2
✓ Branch 2 taken 58 times.
✗ Branch 3 not taken.
 58 WritePipe(fd, path.GetChars(), size * sizeof(char));
112 58 }
113
114 29 void BlockingSend(int fd, const std::string &string) const {
115 29 const size_t size = string.size();
116
1/2
✓ Branch 1 taken 29 times.
✗ Branch 2 not taken.
 29 BlockingSend(fd, size);
117
1/2
✓ Branch 2 taken 29 times.
✗ Branch 3 not taken.
 29 WritePipe(fd, string.data(), size * sizeof(char));
118 29 }
119
120 template<typename CT,
121 typename = typename std::enable_if<
122 std::is_trivially_copyable<CT>::value>::type>
123 696 CT BlockingReceive(int fd) const {
124 typedef typename std::remove_cv<CT>::type T;
125 static_assert(
126 sizeof(T) <= PIPE_BUF,
127 "Type too big to be guaranteed atomic transmission over a pipe");
128 CT item;
129
1/2
✓ Branch 1 taken 348 times.
✗ Branch 2 not taken.
 696 ReadPipe(fd, static_cast<void *>(&item), sizeof(CT));
130 696 return item;
131 }
132
133 319 std::string BlockingReceive(int fd) const {
134 319 const size_t size = BlockingReceive<size_t>(fd);
135
1/2
✓ Branch 2 taken 319 times.
✗ Branch 3 not taken.
 319 std::string result(size, '\t');
136
2/4
✓ Branch 1 taken 319 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 319 times.
✗ Branch 5 not taken.
 319 ReadPipe(fd, static_cast<void *>(&result[0]), size * sizeof(char));
137 319 return result;
138 }
139
140 MountPoint *mount_point_;
141
142 // Pool of fetcher threads. All workers share pipe_bm_[0] (read end)
143 // and serialize their reads via worker_read_mutex_ so cmd+payload
144 // pairs are received atomically.
145 std::vector<std::unique_ptr<pthread_t> > fetcher_threads_;
146 std::unique_ptr<pthread_t> dispatcher_thread_;
147 pthread_mutex_t worker_read_mutex_;
148 size_t pool_size_;
149
150 enum class Command {
151 kTerminate,
152 kFetch
153 };
154
155 /**
156 * Dependency work queue (a pipe with a non-blocking write end). The
157 * dispatcher writes Command + path payload to pipe_bm_[1]; workers read
158 * from pipe_bm_[0] under worker_read_mutex_.
159 */
160 int pipe_bm_[2];
161 /**
162 * Trigger queue (a pipe with a non-blocking write end). ScheduleTrigger()
163 * writes Command + path payload to pipe_triggers_[1]; the dispatcher
164 * reads from pipe_triggers_[0].
165 */
166 int pipe_triggers_[2];
167 /**
168 * Set on destruction: queued work is drained but no longer processed, so
169 * that unmounting does not wait for pending downloads.
170 */
171 atomic_int32 terminating_;
172 bool is_valid_ = true;
173 };
174 #endif // CVMFS_BUNDLE_MGR_H_
175