GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/catalog_traversal_parallel.h
Date: 2026-10-18 02:40:11
Exec Total Coverage
Lines: 187 199 94.0%
Branches: 160 244 65.6%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #ifndef CVMFS_CATALOG_TRAVERSAL_PARALLEL_H_
6 #define CVMFS_CATALOG_TRAVERSAL_PARALLEL_H_
7
8 #include <atomic>
9 #include <stack>
10 #include <string>
11 #include <vector>
12
13 #include "catalog_traversal.h"
14 #include "util/exception.h"
15 #include "util/tube.h"
16
17 namespace swissknife {
18
19 /**
20 * This class implements the same functionality as CatalogTraversal, but in
21 * parallel. For common functionality, see the documentation of
22 * CatalogTraversal. Differences:
23 * - can choose number of threads
24 * - traversal types change meaning:
25 * - depth-first -> parallelized post-order traversal (parents are processed
26 * after all children are finished)
27 * - breadth-first -> same as original, but parallelized
28 */
29 template<class ObjectFetcherT>
30 class CatalogTraversalParallel : public CatalogTraversalBase<ObjectFetcherT> {
31 public:
32 typedef CatalogTraversalBase<ObjectFetcherT> Base;
33 typedef ObjectFetcherT ObjectFetcherTN;
34 typedef typename ObjectFetcherT::CatalogTN CatalogTN;
35 typedef typename ObjectFetcherT::HistoryTN HistoryTN;
36 typedef CatalogTraversalData<CatalogTN> CallbackDataTN;
37 typedef typename CatalogTN::NestedCatalogList NestedCatalogList;
38 typedef typename Base::Parameters Parameters;
39 typedef typename Base::TraversalType TraversalType;
40 typedef std::vector<shash::Any> HashList;
41
42 1859 explicit CatalogTraversalParallel(const Parameters &params)
43 : CatalogTraversalBase<ObjectFetcherT>(params)
44 1859 , num_threads_(params.num_threads)
45
4/8
✓ Branch 2 taken 1859 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 1859 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 1859 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 1859 times.
✗ Branch 12 not taken.
 1859 , serialize_callbacks_(params.serialize_callbacks) {
46 1859 num_errors_.store(0);
47
1/2
✓ Branch 1 taken 1859 times.
✗ Branch 2 not taken.
 1859 shash::Any null_hash;
48 1859 null_hash.SetNull();
49
1/2
✓ Branch 1 taken 1859 times.
✗ Branch 2 not taken.
 1859 catalogs_processing_.Init(1024, null_hash, hasher);
50
1/2
✓ Branch 1 taken 1859 times.
✗ Branch 2 not taken.
 1859 catalogs_done_.Init(1024, null_hash, hasher);
51 1859 pthread_mutex_init(&catalog_callback_lock_, NULL);
52 1859 pthread_mutex_init(&catalogs_lock_, NULL);
53 1859 effective_history_depth_ = this->default_history_depth_;
54 1859 effective_timestamp_threshold_ = this->default_timestamp_threshold_;
55 1859 }
56
57 protected:
58 struct CatalogJob : public CatalogTraversal<ObjectFetcherT>::CatalogJob,
59 public Observable<int> {
60 12158729 explicit CatalogJob(const std::string &path,
61 const shash::Any &hash,
62 const unsigned tree_level,
63 const uint64_t history_depth,
64 CatalogTN *parent = NULL)
65 : CatalogTraversal<ObjectFetcherT>::CatalogJob(path, hash, tree_level,
66 12158729 history_depth, parent) {
67 12158729 children_unprocessed.store(0);
68 12158729 }
69
70
1/2
✓ Branch 1 taken 6061801 times.
✗ Branch 2 not taken.
 6061966 void WakeParents() { this->NotifyListeners(0); }
71
72 std::atomic<int32_t> children_unprocessed;
73 };
74
75 public:
76 /**
77 * Starts the traversal process.
78 * After calling this methods CatalogTraversal will go through all catalogs
79 * and call the registered callback methods for each found catalog.
80 * If something goes wrong in the process, the traversal will be cancelled.
81 *
82 * @return true, when all catalogs were successfully processed. On
83 * failure the traversal is cancelled and false is returned.
84 */
85 1299 bool Traverse(const TraversalType type = Base::kBreadthFirst) {
86
1/2
✓ Branch 1 taken 1299 times.
✗ Branch 2 not taken.
 1299 const shash::Any root_catalog_hash = this->GetRepositoryRootCatalogHash();
87
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 1299 times.
 1299 if (root_catalog_hash.IsNull()) {
88 ✗ return false;
89 }
90
1/2
✓ Branch 1 taken 1299 times.
✗ Branch 2 not taken.
 1299 return Traverse(root_catalog_hash, type);
91 }
92
93 /**
94 * Starts the traversal process at the catalog pointed to by the given hash
95 *
96 * @param root_catalog_hash the entry point into the catalog traversal
97 * @return true when catalogs were successfully traversed
98 */
99 1794 bool Traverse(const shash::Any &root_catalog_hash,
100 const TraversalType type = Base::kBreadthFirst) {
101 // add the root catalog of the repository as the first element on the job
102 // stack
103 3588 if (this->no_repeat_history_
104
4/6
✓ Branch 0 taken 1068 times.
✓ Branch 1 taken 726 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1068 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 1794 times.
 1794 && catalogs_done_.Contains(root_catalog_hash)) {
105 ✗ return true;
106 }
107 1794 effective_traversal_type_ = type;
108
3/6
✓ Branch 2 taken 1794 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 1794 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 1794 times.
✗ Branch 9 not taken.
 1794 CatalogJob *root_job = new CatalogJob("", root_catalog_hash, 0, 0);
109 1794 PushJob(root_job);
110 1794 return DoTraverse();
111 }
112
113 /**
114 * Start the traversal process from a list of root catalogs. Same as
115 * TraverseRevision function, TraverseList does not traverse into predecessor
116 * catalog revisions and ignores TraversalParameter settings.
117 */
118 1575 bool TraverseList(const HashList &root_catalog_list,
119 const TraversalType type = Base::kBreadthFirst) {
120 // Push in reverse order for CatalogTraversal-like behavior
121 1575 HashList::const_reverse_iterator i = root_catalog_list.rbegin();
122 1575 const HashList::const_reverse_iterator iend = root_catalog_list.rend();
123 1575 bool has_pushed = false;
124 {
125 1575 MutexLockGuard const m(&catalogs_lock_);
126
3/4
✓ Branch 2 taken 4796 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 3221 times.
✓ Branch 5 taken 1575 times.
 4796 for (; i != iend; ++i) {
127
7/8
✓ Branch 0 taken 2924 times.
✓ Branch 1 taken 297 times.
✓ Branch 4 taken 2924 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 1088 times.
✓ Branch 7 taken 1836 times.
✓ Branch 8 taken 1088 times.
✓ Branch 9 taken 2133 times.
 3221 if (this->no_repeat_history_ && catalogs_done_.Contains(*i)) {
128 1088 continue;
129 }
130
131
3/6
✓ Branch 2 taken 2133 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 2133 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 2133 times.
✗ Branch 10 not taken.
 2133 CatalogJob *root_job = new CatalogJob("", *i, 0, 0);
132
1/2
✓ Branch 1 taken 2133 times.
✗ Branch 2 not taken.
 2133 PushJobUnlocked(root_job);
133 2133 has_pushed = true;
134 }
135 1575 }
136 // noop: no catalogs to traverse
137
2/2
✓ Branch 0 taken 416 times.
✓ Branch 1 taken 1159 times.
 1575 if (!has_pushed) {
138 416 return true;
139 }
140 1159 effective_traversal_type_ = type;
141 1159 effective_history_depth_ = Parameters::kNoHistory;
142 1159 effective_timestamp_threshold_ = Parameters::kNoTimestampThreshold;
143
1/2
✓ Branch 1 taken 1159 times.
✗ Branch 2 not taken.
 1159 bool const result = DoTraverse();
144 1159 effective_history_depth_ = this->default_history_depth_;
145 1159 effective_timestamp_threshold_ = this->default_timestamp_threshold_;
146 1159 return result;
147 }
148
149 /**
150 * Starts the traversal process at the catalog pointed to by the given hash
151 * but doesn't traverse into predecessor catalog revisions. This overrides the
152 * TraversalParameter settings provided at construction.
153 *
154 * @param root_catalog_hash the entry point into the catalog traversal
155 * @return true when catalogs were successfully traversed
156 */
157 66 bool TraverseRevision(const shash::Any &root_catalog_hash,
158 const TraversalType type = Base::kBreadthFirst) {
159 66 effective_history_depth_ = Parameters::kNoHistory;
160 66 effective_timestamp_threshold_ = Parameters::kNoTimestampThreshold;
161 66 bool const result = Traverse(root_catalog_hash, type);
162 66 effective_history_depth_ = this->default_history_depth_;
163 66 effective_timestamp_threshold_ = this->default_timestamp_threshold_;
164 66 return result;
165 }
166
167 protected:
168 55606084 static uint32_t hasher(const shash::Any &key) {
169 // Don't start with the first bytes, because == is using them as well
170 return static_cast<uint32_t>(
171 55606084 *(reinterpret_cast<const uint32_t *>(key.digest) + 1));
172 }
173
174 2953 bool DoTraverse() {
175 // Optimal number of threads is yet to be determined. The main event loop
176 // contains a spin-lock, so it should not be more than number of cores.
177 2953 threads_process_ = reinterpret_cast<pthread_t *>(
178 2953 smalloc(sizeof(pthread_t) * num_threads_));
179
2/2
✓ Branch 0 taken 3415 times.
✓ Branch 1 taken 2953 times.
 6368 for (unsigned int i = 0; i < num_threads_; ++i) {
180 3415 int const retval = pthread_create(&threads_process_[i], NULL,
181 MainProcessQueue, this);
182
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3415 times.
 3415 if (retval != 0)
183 ✗ PANIC(kLogStderr, "failed to create thread");
184 }
185
186
2/2
✓ Branch 0 taken 3415 times.
✓ Branch 1 taken 2953 times.
 6368 for (unsigned int i = 0; i < num_threads_; ++i) {
187 3415 int const retval = pthread_join(threads_process_[i], NULL);
188
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3415 times.
 3415 assert(retval == 0);
189 }
190 2953 free(threads_process_);
191
192
2/2
✓ Branch 1 taken 33 times.
✓ Branch 2 taken 2920 times.
 5906 if (num_errors_.load())
193 33 return false;
194
195
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2920 times.
 2920 assert(catalogs_processing_.size() == 0);
196
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2920 times.
 2920 assert(pre_job_queue_.IsEmpty());
197
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2920 times.
 2920 assert(post_job_queue_.IsEmpty());
198 2920 return true;
199 }
200
201 3415 static void *MainProcessQueue(void *data) {
202 3415 CatalogTraversalParallel<ObjectFetcherT> *traversal = reinterpret_cast<
203 CatalogTraversalParallel<ObjectFetcherT> *>(data);
204 CatalogJob *current_job;
205 while (true) {
206 13277039 current_job = traversal->post_job_queue_.TryPopFront();
207
2/2
✓ Branch 0 taken 1121346 times.
✓ Branch 1 taken 12157838 times.
 13279184 if (current_job != NULL) {
208 1121346 traversal->ProcessJobPost(current_job);
209 } else {
210 12157838 current_job = traversal->pre_job_queue_.PopFront();
211 // NULL means the master thread tells us to finish
212
2/2
✓ Branch 1 taken 3382 times.
✓ Branch 2 taken 12154720 times.
 12158300 if (current_job->hash.IsNull()) {
213
1/2
✓ Branch 0 taken 3382 times.
✗ Branch 1 not taken.
 3382 delete current_job;
214 3415 break;
215 }
216 12154720 traversal->ProcessJobPre(current_job);
217 }
218 }
219 3415 return NULL;
220 }
221
222 2953 void NotifyFinished() {
223
1/2
✓ Branch 1 taken 2953 times.
✗ Branch 2 not taken.
 2953 shash::Any null_hash;
224 2953 null_hash.SetNull();
225
2/2
✓ Branch 0 taken 3415 times.
✓ Branch 1 taken 2953 times.
 6368 for (unsigned i = 0; i < num_threads_; ++i) {
226
3/6
✓ Branch 2 taken 3415 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 3415 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 3415 times.
✗ Branch 9 not taken.
 3415 CatalogJob *job = new CatalogJob("", null_hash, 0, 0);
227
1/2
✓ Branch 1 taken 3415 times.
✗ Branch 2 not taken.
 3415 pre_job_queue_.EnqueueFront(job);
228 }
229 2953 }
230
231 1794 void PushJob(CatalogJob *job) {
232 1794 MutexLockGuard const m(&catalogs_lock_);
233
1/2
✓ Branch 1 taken 1794 times.
✗ Branch 2 not taken.
 1794 PushJobUnlocked(job);
234 1794 }
235
236 12155314 void PushJobUnlocked(CatalogJob *job) {
237 12155314 catalogs_processing_.Insert(job->hash, job);
238 12155314 pre_job_queue_.EnqueueFront(job);
239 12155314 }
240
241 12153763 void ProcessJobPre(CatalogJob *job) {
242
4/6
✓ Branch 0 taken 12153763 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 12151321 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 33 times.
✓ Branch 6 taken 12151288 times.
 12153763 if (!this->PrepareCatalog(job)) {
243 33 num_errors_.fetch_add(1);
244
1/2
✓ Branch 1 taken 33 times.
✗ Branch 2 not taken.
 33 NotifyFinished();
245 11031262 return;
246 }
247
2/2
✓ Branch 0 taken 195 times.
✓ Branch 1 taken 12151093 times.
 12151288 if (job->ignore) {
248
1/2
✓ Branch 1 taken 195 times.
✗ Branch 2 not taken.
 195 FinalizeJob(job);
249 195 return;
250 }
251
1/2
✓ Branch 1 taken 12126508 times.
✗ Branch 2 not taken.
 12151093 NestedCatalogList const catalog_list = job->catalog
252 ->ListOwnNestedCatalogs();
253 unsigned int num_children;
254 // Ensure that pushed children won't call ProcessJobPost on this job
255 // before this function finishes
256 {
257 12126508 MutexLockGuard const m(&catalogs_lock_);
258
2/2
✓ Branch 0 taken 6092953 times.
✓ Branch 1 taken 6062100 times.
 12155053 if (effective_traversal_type_ == Base::kBreadthFirst) {
259
1/2
✓ Branch 1 taken 6092953 times.
✗ Branch 2 not taken.
 6092953 num_children = PushPreviousRevision(job)
260
1/2
✓ Branch 1 taken 6092953 times.
✗ Branch 2 not taken.
 6092953 + PushNestedCatalogs(job, catalog_list);
261 } else {
262
1/2
✓ Branch 1 taken 6062100 times.
✗ Branch 2 not taken.
 6062100 num_children = PushNestedCatalogs(job, catalog_list)
263
1/2
✓ Branch 1 taken 6062100 times.
✗ Branch 2 not taken.
 6062100 + PushPreviousRevision(job);
264 6062100 job->children_unprocessed.store(num_children);
265 }
266
3/6
✓ Branch 0 taken 12155053 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 12155053 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 12155053 times.
 12155053 if (!this->CloseCatalog(false, job)) {
267 ✗ num_errors_.fetch_add(1);
268 ✗ NotifyFinished();
269 }
270 12155053 }
271
272 // breadth-first: can post-process immediately
273 // depth-first: no children -> can post-process immediately
274
4/4
✓ Branch 0 taken 6062067 times.
✓ Branch 1 taken 6092788 times.
✓ Branch 2 taken 4940688 times.
✓ Branch 3 taken 1121379 times.
 12154855 if (effective_traversal_type_ == Base::kBreadthFirst || num_children == 0) {
275
1/2
✓ Branch 1 taken 11030572 times.
✗ Branch 2 not taken.
 11033476 ProcessJobPost(job);
276 11030572 return;
277 }
278
2/2
✓ Branch 1 taken 1121379 times.
✓ Branch 2 taken 11031034 times.
 12151951 }
279
280 12155053 unsigned int PushNestedCatalogs(CatalogJob *job,
281 const NestedCatalogList &catalog_list) {
282 12155053 typename NestedCatalogList::const_iterator i = catalog_list.begin();
283 12155053 typename NestedCatalogList::const_iterator const iend = catalog_list.end();
284 12155053 unsigned int num_children = 0;
285
2/2
✓ Branch 2 taken 12151335 times.
✓ Branch 3 taken 12155053 times.
 24306388 for (; i != iend; ++i) {
286
7/8
✓ Branch 0 taken 20799 times.
✓ Branch 1 taken 12130536 times.
✓ Branch 4 taken 20799 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2199 times.
✓ Branch 7 taken 18600 times.
✓ Branch 8 taken 2199 times.
✓ Branch 9 taken 12149136 times.
 12151335 if (this->no_repeat_history_ && catalogs_done_.Contains(i->hash)) {
287 2199 continue;
288 }
289
290 CatalogJob *child;
291 24298272 if (!this->no_repeat_history_
292
7/8
✓ Branch 0 taken 18600 times.
✓ Branch 1 taken 12130536 times.
✓ Branch 4 taken 18600 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 18336 times.
✓ Branch 7 taken 264 times.
✓ Branch 8 taken 12148872 times.
✓ Branch 9 taken 264 times.
 12149136 || !catalogs_processing_.Lookup(i->hash, &child)) {
293
2/2
✓ Branch 0 taken 12099021 times.
✓ Branch 1 taken 49851 times.
 12148872 CatalogTN *parent = (this->no_close_) ? job->catalog : NULL;
294
1/2
✓ Branch 2 taken 12148872 times.
✗ Branch 3 not taken.
 24297744 child = new CatalogJob(i->mountpoint.ToString(),
295
1/2
✓ Branch 2 taken 12148872 times.
✗ Branch 3 not taken.
 12148872 i->hash,
296 12148872 job->tree_level + 1,
297
1/2
✓ Branch 1 taken 12148872 times.
✗ Branch 2 not taken.
 12148872 job->history_depth,
298 parent);
299
1/2
✓ Branch 1 taken 12148872 times.
✗ Branch 2 not taken.
 12148872 PushJobUnlocked(child);
300 }
301
302
2/2
✓ Branch 0 taken 6060738 times.
✓ Branch 1 taken 6088398 times.
 12149136 if (effective_traversal_type_ == Base::kDepthFirst) {
303
1/2
✓ Branch 1 taken 6060738 times.
✗ Branch 2 not taken.
 6060738 child->RegisterListener(&CatalogTraversalParallel::OnChildFinished,
304 this, job);
305 }
306 12149136 ++num_children;
307 }
308 12155053 return num_children;
309 }
310
311 /**
312 * Pushes the previous revision of a root catalog.
313 * @return the number of catalogs pushed on the processing stack
314 */
315 12155053 unsigned int PushPreviousRevision(CatalogJob *job) {
316 // only root catalogs are used for entering a previous revision (graph)
317
2/2
✓ Branch 1 taken 12148774 times.
✓ Branch 2 taken 6279 times.
 12155053 if (!job->catalog->IsRoot()) {
318 12148774 return 0;
319 }
320
321
1/2
✓ Branch 1 taken 6279 times.
✗ Branch 2 not taken.
 6279 const shash::Any previous_revision = job->catalog->GetPreviousRevision();
322
2/2
✓ Branch 1 taken 615 times.
✓ Branch 2 taken 5664 times.
 6279 if (previous_revision.IsNull()) {
323 615 return 0;
324 }
325
326 // check if the next deeper history level is actually requested
327
3/4
✓ Branch 1 taken 5664 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 3149 times.
✓ Branch 4 taken 2515 times.
 5664 if (this->IsBelowPruningThresholds(*job, effective_history_depth_,
328 effective_timestamp_threshold_)) {
329 3149 return 0;
330 }
331
332 5030 if (this->no_repeat_history_
333
5/8
✓ Branch 0 taken 1591 times.
✓ Branch 1 taken 924 times.
✓ Branch 3 taken 1591 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 1591 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 2515 times.
 2515 && catalogs_done_.Contains(previous_revision)) {
334 ✗ return 0;
335 }
336
337 CatalogJob *prev_job;
338 5030 if (!this->no_repeat_history_
339
5/8
✓ Branch 0 taken 1591 times.
✓ Branch 1 taken 924 times.
✓ Branch 3 taken 1591 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 1591 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 2515 times.
✗ Branch 8 not taken.
 2515 || !catalogs_processing_.Lookup(previous_revision, &prev_job)) {
340
2/4
✓ Branch 2 taken 2515 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2515 times.
✗ Branch 6 not taken.
 5030 prev_job = new CatalogJob("", previous_revision, 0,
341
1/2
✓ Branch 1 taken 2515 times.
✗ Branch 2 not taken.
 2515 job->history_depth + 1);
342
1/2
✓ Branch 1 taken 2515 times.
✗ Branch 2 not taken.
 2515 PushJobUnlocked(prev_job);
343 }
344
345
2/2
✓ Branch 0 taken 495 times.
✓ Branch 1 taken 2020 times.
 2515 if (effective_traversal_type_ == Base::kDepthFirst) {
346
1/2
✓ Branch 1 taken 495 times.
✗ Branch 2 not taken.
 495 prev_job->RegisterListener(&CatalogTraversalParallel::OnChildFinished,
347 this, job);
348 }
349 2515 return 1;
350 }
351
352 12153634 void ProcessJobPost(CatalogJob *job) {
353 // Save time by keeping catalog open when suitable
354
1/2
✓ Branch 0 taken 12153634 times.
✗ Branch 1 not taken.
 12153634 if (job->catalog == NULL) {
355
2/4
✓ Branch 0 taken 12153766 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 12128686 times.
 12153634 if (!this->ReopenCatalog(job)) {
356 ✗ num_errors_.fetch_add(1);
357 ✗ NotifyFinished();
358 ✗ return;
359 }
360 }
361
2/2
✓ Branch 0 taken 12127861 times.
✓ Branch 1 taken 825 times.
 12128686 if (serialize_callbacks_) {
362 12127861 MutexLockGuard const m(&catalog_callback_lock_);
363
2/4
✓ Branch 1 taken 12155053 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 12155053 times.
✗ Branch 5 not taken.
 12155053 this->NotifyListeners(job->GetCallbackData());
364 12155053 } else {
365
0/2
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 825 this->NotifyListeners(job->GetCallbackData());
366 }
367
2/2
✓ Branch 0 taken 55867 times.
✓ Branch 1 taken 12099153 times.
 12155020 if (!this->no_close_) {
368
2/4
✓ Branch 0 taken 55867 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 55867 times.
 55867 if (!this->CloseCatalog(true, job)) {
369 ✗ num_errors_.fetch_add(1);
370 ✗ NotifyFinished();
371 ✗ return;
372 }
373 }
374 12155020 FinalizeJob(job);
375 }
376
377 12155182 void FinalizeJob(CatalogJob *job) {
378 {
379 12155182 MutexLockGuard const m(&catalogs_lock_);
380
1/2
✓ Branch 1 taken 12155248 times.
✗ Branch 2 not taken.
 12155248 catalogs_processing_.Erase(job->hash);
381
1/2
✓ Branch 1 taken 12155248 times.
✗ Branch 2 not taken.
 12155248 catalogs_done_.Insert(job->hash, true);
382 // No more catalogs to process -> finish
383
1/2
✓ Branch 2 taken 2920 times.
✗ Branch 3 not taken.
 12158168 if (catalogs_processing_.size() == 0 && pre_job_queue_.IsEmpty()
384
5/6
✓ Branch 0 taken 2920 times.
✓ Branch 1 taken 12152328 times.
✓ Branch 3 taken 2920 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 2920 times.
✓ Branch 6 taken 12152328 times.
 12158168 && post_job_queue_.IsEmpty()) {
385
1/2
✓ Branch 1 taken 2920 times.
✗ Branch 2 not taken.
 2920 NotifyFinished();
386 }
387 12155248 }
388
2/2
✓ Branch 0 taken 6061966 times.
✓ Branch 1 taken 6093150 times.
 12155116 if (effective_traversal_type_ == Base::kDepthFirst) {
389 6061966 job->WakeParents();
390 }
391
2/2
✓ Branch 0 taken 12154621 times.
✓ Branch 1 taken 231 times.
 12154852 delete job;
392 12153004 }
393
394 6060804 void OnChildFinished(const int &a, CatalogJob *job) {
395 // atomic_xadd32 returns value before subtraction -> needs to equal 1
396
2/2
✓ Branch 0 taken 1121379 times.
✓ Branch 1 taken 4939425 times.
 12121608 if (job->children_unprocessed.fetch_add(-1) == 1) {
397 1121379 post_job_queue_.EnqueueFront(job);
398 }
399 6060804 }
400
401 unsigned int num_threads_;
402 bool serialize_callbacks_;
403
404 uint64_t effective_history_depth_;
405 time_t effective_timestamp_threshold_;
406 TraversalType effective_traversal_type_;
407
408 pthread_t *threads_process_;
409 std::atomic<int32_t> num_errors_;
410
411 Tube<CatalogJob> pre_job_queue_;
412 Tube<CatalogJob> post_job_queue_;
413 SmallHashDynamic<shash::Any, CatalogJob *> catalogs_processing_;
414 SmallHashDynamic<shash::Any, bool> catalogs_done_;
415 pthread_mutex_t catalogs_lock_;
416
417 pthread_mutex_t catalog_callback_lock_;
418 };
419
420 } // namespace swissknife
421
422 #endif // CVMFS_CATALOG_TRAVERSAL_PARALLEL_H_
423