GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/catalog_traversal_parallel.h
Date: 2026-08-16 02:40:26
Exec Total Coverage
Lines: 186 199 93.5%
Branches: 158 244 64.8%
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 <stack>
9 #include <string>
10 #include <vector>
11
12 #include "catalog_traversal.h"
13 #include "util/atomic.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 2265 explicit CatalogTraversalParallel(const Parameters &params)
43 : CatalogTraversalBase<ObjectFetcherT>(params)
44 2265 , num_threads_(params.num_threads)
45
4/8
✓ Branch 2 taken 2265 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2265 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2265 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 2265 times.
✗ Branch 12 not taken.
 2265 , serialize_callbacks_(params.serialize_callbacks) {
46 2265 atomic_init32(&num_errors_);
47
1/2
✓ Branch 1 taken 2265 times.
✗ Branch 2 not taken.
 2265 shash::Any null_hash;
48 2265 null_hash.SetNull();
49
1/2
✓ Branch 1 taken 2265 times.
✗ Branch 2 not taken.
 2265 catalogs_processing_.Init(1024, null_hash, hasher);
50
1/2
✓ Branch 1 taken 2265 times.
✗ Branch 2 not taken.
 2265 catalogs_done_.Init(1024, null_hash, hasher);
51 2265 pthread_mutex_init(&catalog_callback_lock_, NULL);
52 2265 pthread_mutex_init(&catalogs_lock_, NULL);
53 2265 effective_history_depth_ = this->default_history_depth_;
54 2265 effective_timestamp_threshold_ = this->default_timestamp_threshold_;
55 2265 }
56
57 protected:
58 struct CatalogJob : public CatalogTraversal<ObjectFetcherT>::CatalogJob,
59 public Observable<int> {
60 18045945 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 18045945 history_depth, parent) {
67 18045945 atomic_init32(&children_unprocessed);
68 18045945 }
69
70
1/2
✓ Branch 1 taken 8999178 times.
✗ Branch 2 not taken.
 8999619 void WakeParents() { this->NotifyListeners(0); }
71
72 atomic_int32 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 1456 bool Traverse(const TraversalType type = Base::kBreadthFirst) {
86
1/2
✓ Branch 1 taken 1456 times.
✗ Branch 2 not taken.
 1456 const shash::Any root_catalog_hash = this->GetRepositoryRootCatalogHash();
87
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 1456 times.
 1456 if (root_catalog_hash.IsNull()) {
88 ✗ return false;
89 }
90
1/2
✓ Branch 1 taken 1456 times.
✗ Branch 2 not taken.
 1456 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 2191 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 4382 if (this->no_repeat_history_
104
4/6
✓ Branch 0 taken 1113 times.
✓ Branch 1 taken 1078 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1113 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2191 times.
 2191 && catalogs_done_.Contains(root_catalog_hash)) {
105 ✗ return true;
106 }
107 2191 effective_traversal_type_ = type;
108
3/6
✓ Branch 2 taken 2191 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2191 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2191 times.
✗ Branch 9 not taken.
 2191 CatalogJob *root_job = new CatalogJob("", root_catalog_hash, 0, 0);
109 2191 PushJob(root_job);
110 2191 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 1393 bool TraverseList(const HashList &root_catalog_list,
119 const TraversalType type = Base::kBreadthFirst) {
120 // Push in reverse order for CatalogTraversal-like behavior
121 1393 HashList::const_reverse_iterator i = root_catalog_list.rbegin();
122 1393 const HashList::const_reverse_iterator iend = root_catalog_list.rend();
123 1393 bool has_pushed = false;
124 {
125 1393 MutexLockGuard const m(&catalogs_lock_);
126
3/4
✓ Branch 2 taken 4397 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 3004 times.
✓ Branch 5 taken 1393 times.
 4397 for (; i != iend; ++i) {
127
7/8
✓ Branch 0 taken 2563 times.
✓ Branch 1 taken 441 times.
✓ Branch 4 taken 2563 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 850 times.
✓ Branch 7 taken 1713 times.
✓ Branch 8 taken 850 times.
✓ Branch 9 taken 2154 times.
 3004 if (this->no_repeat_history_ && catalogs_done_.Contains(*i)) {
128 850 continue;
129 }
130
131
3/6
✓ Branch 2 taken 2154 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 2154 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 2154 times.
✗ Branch 10 not taken.
 2154 CatalogJob *root_job = new CatalogJob("", *i, 0, 0);
132
1/2
✓ Branch 1 taken 2154 times.
✗ Branch 2 not taken.
 2154 PushJobUnlocked(root_job);
133 2154 has_pushed = true;
134 }
135 1393 }
136 // noop: no catalogs to traverse
137
2/2
✓ Branch 0 taken 325 times.
✓ Branch 1 taken 1068 times.
 1393 if (!has_pushed) {
138 325 return true;
139 }
140 1068 effective_traversal_type_ = type;
141 1068 effective_history_depth_ = Parameters::kNoHistory;
142 1068 effective_timestamp_threshold_ = Parameters::kNoTimestampThreshold;
143
1/2
✓ Branch 1 taken 1068 times.
✗ Branch 2 not taken.
 1068 bool const result = DoTraverse();
144 1068 effective_history_depth_ = this->default_history_depth_;
145 1068 effective_timestamp_threshold_ = this->default_timestamp_threshold_;
146 1068 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 98 bool TraverseRevision(const shash::Any &root_catalog_hash,
158 const TraversalType type = Base::kBreadthFirst) {
159 98 effective_history_depth_ = Parameters::kNoHistory;
160 98 effective_timestamp_threshold_ = Parameters::kNoTimestampThreshold;
161 98 bool const result = Traverse(root_catalog_hash, type);
162 98 effective_history_depth_ = this->default_history_depth_;
163 98 effective_timestamp_threshold_ = this->default_timestamp_threshold_;
164 98 return result;
165 }
166
167 protected:
168 82586962 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 82586962 *(reinterpret_cast<const uint32_t *>(key.digest) + 1));
172 }
173
174 3259 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 3259 threads_process_ = reinterpret_cast<pthread_t *>(
178 3259 smalloc(sizeof(pthread_t) * num_threads_));
179
2/2
✓ Branch 0 taken 3945 times.
✓ Branch 1 taken 3259 times.
 7204 for (unsigned int i = 0; i < num_threads_; ++i) {
180 3945 int const retval = pthread_create(&threads_process_[i], NULL, MainProcessQueue,
181 this);
182
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3945 times.
 3945 if (retval != 0)
183 ✗ PANIC(kLogStderr, "failed to create thread");
184 }
185
186
2/2
✓ Branch 0 taken 3945 times.
✓ Branch 1 taken 3259 times.
 7204 for (unsigned int i = 0; i < num_threads_; ++i) {
187 3945 int const retval = pthread_join(threads_process_[i], NULL);
188
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3945 times.
 3945 assert(retval == 0);
189 }
190 3259 free(threads_process_);
191
192
2/2
✓ Branch 1 taken 49 times.
✓ Branch 2 taken 3210 times.
 3259 if (atomic_read32(&num_errors_))
193 49 return false;
194
195
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 3210 times.
 3210 assert(catalogs_processing_.size() == 0);
196
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 3210 times.
 3210 assert(pre_job_queue_.IsEmpty());
197
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 3210 times.
 3210 assert(post_job_queue_.IsEmpty());
198 3210 return true;
199 }
200
201 3945 static void *MainProcessQueue(void *data) {
202 3945 CatalogTraversalParallel<ObjectFetcherT> *traversal = reinterpret_cast<
203 CatalogTraversalParallel<ObjectFetcherT> *>(data);
204 CatalogJob *current_job;
205 while (true) {
206 19707965 current_job = traversal->post_job_queue_.TryPopFront();
207
2/2
✓ Branch 0 taken 1664470 times.
✓ Branch 1 taken 18044769 times.
 19709239 if (current_job != NULL) {
208 1664470 traversal->ProcessJobPost(current_job);
209 } else {
210 18044769 current_job = traversal->pre_job_queue_.PopFront();
211 // NULL means the master thread tells us to finish
212
2/2
✓ Branch 1 taken 3945 times.
✓ Branch 2 taken 18041657 times.
 18045700 if (current_job->hash.IsNull()) {
213
1/2
✓ Branch 0 taken 3945 times.
✗ Branch 1 not taken.
 3945 delete current_job;
214 3945 break;
215 }
216 18041657 traversal->ProcessJobPre(current_job);
217 }
218 }
219 3945 return NULL;
220 }
221
222 3259 void NotifyFinished() {
223
1/2
✓ Branch 1 taken 3259 times.
✗ Branch 2 not taken.
 3259 shash::Any null_hash;
224 3259 null_hash.SetNull();
225
2/2
✓ Branch 0 taken 3945 times.
✓ Branch 1 taken 3259 times.
 7204 for (unsigned i = 0; i < num_threads_; ++i) {
226
3/6
✓ Branch 2 taken 3945 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 3945 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 3945 times.
✗ Branch 9 not taken.
 3945 CatalogJob *job = new CatalogJob("", null_hash, 0, 0);
227
1/2
✓ Branch 1 taken 3945 times.
✗ Branch 2 not taken.
 3945 pre_job_queue_.EnqueueFront(job);
228 }
229 3259 }
230
231 2191 void PushJob(CatalogJob *job) {
232 2191 MutexLockGuard const m(&catalogs_lock_);
233
1/2
✓ Branch 1 taken 2191 times.
✗ Branch 2 not taken.
 2191 PushJobUnlocked(job);
234 2191 }
235
236 18042000 void PushJobUnlocked(CatalogJob *job) {
237 18042000 catalogs_processing_.Insert(job->hash, job);
238 18042000 pre_job_queue_.EnqueueFront(job);
239 18042000 }
240
241 18041608 void ProcessJobPre(CatalogJob *job) {
242
4/6
✓ Branch 0 taken 18041608 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 18038521 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 49 times.
✓ Branch 6 taken 18038472 times.
 18041608 if (!this->PrepareCatalog(job)) {
243 49 atomic_inc32(&num_errors_);
244
1/2
✓ Branch 1 taken 49 times.
✗ Branch 2 not taken.
 49 NotifyFinished();
245 16375570 return;
246 }
247
2/2
✓ Branch 0 taken 222 times.
✓ Branch 1 taken 18038250 times.
 18038472 if (job->ignore) {
248
1/2
✓ Branch 1 taken 222 times.
✗ Branch 2 not taken.
 222 FinalizeJob(job);
249 222 return;
250 }
251
1/2
✓ Branch 1 taken 17975726 times.
✗ Branch 2 not taken.
 18038250 NestedCatalogList const catalog_list = job->catalog->ListOwnNestedCatalogs();
252 unsigned int num_children;
253 // Ensure that pushed children won't call ProcessJobPost on this job
254 // before this function finishes
255 {
256 17975726 MutexLockGuard const m(&catalogs_lock_);
257
2/2
✓ Branch 0 taken 9041866 times.
✓ Branch 1 taken 8999814 times.
 18041680 if (effective_traversal_type_ == Base::kBreadthFirst) {
258
1/2
✓ Branch 1 taken 9041866 times.
✗ Branch 2 not taken.
 9041866 num_children = PushPreviousRevision(job)
259
1/2
✓ Branch 1 taken 9041866 times.
✗ Branch 2 not taken.
 9041866 + PushNestedCatalogs(job, catalog_list);
260 } else {
261
1/2
✓ Branch 1 taken 8999814 times.
✗ Branch 2 not taken.
 8999814 num_children = PushNestedCatalogs(job, catalog_list)
262
1/2
✓ Branch 1 taken 8999814 times.
✗ Branch 2 not taken.
 8999814 + PushPreviousRevision(job);
263 8999814 atomic_write32(&job->children_unprocessed, num_children);
264 }
265
3/6
✓ Branch 0 taken 18041680 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 18041680 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 18041680 times.
 18041680 if (!this->CloseCatalog(false, job)) {
266 ✗ atomic_inc32(&num_errors_);
267 ✗ NotifyFinished();
268 }
269 18041680 }
270
271 // breadth-first: can post-process immediately
272 // depth-first: no children -> can post-process immediately
273
4/4
✓ Branch 0 taken 8999716 times.
✓ Branch 1 taken 9041474 times.
✓ Branch 2 taken 7335246 times.
✓ Branch 3 taken 1664470 times.
 18041190 if (effective_traversal_type_ == Base::kBreadthFirst || num_children == 0) {
274
1/2
✓ Branch 1 taken 16374074 times.
✗ Branch 2 not taken.
 16376720 ProcessJobPost(job);
275 16374074 return;
276 }
277
2/2
✓ Branch 1 taken 1664470 times.
✓ Branch 2 taken 16375299 times.
 18038544 }
278
279 18041680 unsigned int PushNestedCatalogs(CatalogJob *job,
280 const NestedCatalogList &catalog_list) {
281 18041680 typename NestedCatalogList::const_iterator i = catalog_list.begin();
282 18041680 typename NestedCatalogList::const_iterator const iend = catalog_list.end();
283 18041680 unsigned int num_children = 0;
284
2/2
✓ Branch 2 taken 18037893 times.
✓ Branch 3 taken 18041680 times.
 36079573 for (; i != iend; ++i) {
285
7/8
✓ Branch 0 taken 25885 times.
✓ Branch 1 taken 18012008 times.
✓ Branch 4 taken 25885 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2995 times.
✓ Branch 7 taken 22890 times.
✓ Branch 8 taken 2995 times.
✓ Branch 9 taken 18034898 times.
 18037893 if (this->no_repeat_history_ && catalogs_done_.Contains(i->hash)) {
286 2995 continue;
287 }
288
289 CatalogJob *child;
290 36069796 if (!this->no_repeat_history_
291
7/8
✓ Branch 0 taken 22890 times.
✓ Branch 1 taken 18012008 times.
✓ Branch 4 taken 22890 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 22498 times.
✓ Branch 7 taken 392 times.
✓ Branch 8 taken 18034506 times.
✓ Branch 9 taken 392 times.
 18034898 || !catalogs_processing_.Lookup(i->hash, &child)) {
292
2/2
✓ Branch 0 taken 17965213 times.
✓ Branch 1 taken 69293 times.
 18034506 CatalogTN *parent = (this->no_close_) ? job->catalog : NULL;
293
1/2
✓ Branch 2 taken 18034506 times.
✗ Branch 3 not taken.
 36069012 child = new CatalogJob(i->mountpoint.ToString(),
294
1/2
✓ Branch 2 taken 18034506 times.
✗ Branch 3 not taken.
 18034506 i->hash,
295 18034506 job->tree_level + 1,
296
1/2
✓ Branch 1 taken 18034506 times.
✗ Branch 2 not taken.
 18034506 job->history_depth,
297 parent);
298
1/2
✓ Branch 1 taken 18034506 times.
✗ Branch 2 not taken.
 18034506 PushJobUnlocked(child);
299 }
300
301
2/2
✓ Branch 0 taken 8998332 times.
✓ Branch 1 taken 9036566 times.
 18034898 if (effective_traversal_type_ == Base::kDepthFirst) {
302
1/2
✓ Branch 1 taken 8998332 times.
✗ Branch 2 not taken.
 8998332 child->RegisterListener(&CatalogTraversalParallel::OnChildFinished,
303 this, job);
304 }
305 18034898 ++num_children;
306 }
307 18041680 return num_children;
308 }
309
310 /**
311 * Pushes the previous revision of a root catalog.
312 * @return the number of catalogs pushed on the processing stack
313 */
314 18041680 unsigned int PushPreviousRevision(CatalogJob *job) {
315 // only root catalogs are used for entering a previous revision (graph)
316
2/2
✓ Branch 1 taken 18034383 times.
✓ Branch 2 taken 7297 times.
 18041680 if (!job->catalog->IsRoot()) {
317 18034383 return 0;
318 }
319
320
1/2
✓ Branch 1 taken 7297 times.
✗ Branch 2 not taken.
 7297 const shash::Any previous_revision = job->catalog->GetPreviousRevision();
321
2/2
✓ Branch 1 taken 643 times.
✓ Branch 2 taken 6654 times.
 7297 if (previous_revision.IsNull()) {
322 643 return 0;
323 }
324
325 // check if the next deeper history level is actually requested
326
3/4
✓ Branch 1 taken 6654 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 3505 times.
✓ Branch 4 taken 3149 times.
 6654 if (this->IsBelowPruningThresholds(*job, effective_history_depth_,
327 effective_timestamp_threshold_)) {
328 3505 return 0;
329 }
330
331 6298 if (this->no_repeat_history_
332
5/8
✓ Branch 0 taken 1777 times.
✓ Branch 1 taken 1372 times.
✓ Branch 3 taken 1777 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 1777 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 3149 times.
 3149 && catalogs_done_.Contains(previous_revision)) {
333 ✗ return 0;
334 }
335
336 CatalogJob *prev_job;
337 6298 if (!this->no_repeat_history_
338
5/8
✓ Branch 0 taken 1777 times.
✓ Branch 1 taken 1372 times.
✓ Branch 3 taken 1777 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 1777 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 3149 times.
✗ Branch 8 not taken.
 3149 || !catalogs_processing_.Lookup(previous_revision, &prev_job)) {
339
2/4
✓ Branch 2 taken 3149 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 3149 times.
✗ Branch 6 not taken.
 6298 prev_job = new CatalogJob("", previous_revision, 0,
340
1/2
✓ Branch 1 taken 3149 times.
✗ Branch 2 not taken.
 3149 job->history_depth + 1);
341
1/2
✓ Branch 1 taken 3149 times.
✗ Branch 2 not taken.
 3149 PushJobUnlocked(prev_job);
342 }
343
344
2/2
✓ Branch 0 taken 735 times.
✓ Branch 1 taken 2414 times.
 3149 if (effective_traversal_type_ == Base::kDepthFirst) {
345
1/2
✓ Branch 1 taken 735 times.
✗ Branch 2 not taken.
 735 prev_job->RegisterListener(&CatalogTraversalParallel::OnChildFinished,
346 this, job);
347 }
348 3149 return 1;
349 }
350
351 18039328 void ProcessJobPost(CatalogJob *job) {
352 // Save time by keeping catalog open when suitable
353
1/2
✓ Branch 0 taken 18039377 times.
✗ Branch 1 not taken.
 18039328 if (job->catalog == NULL) {
354
2/4
✓ Branch 0 taken 18039377 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 17973570 times.
 18039377 if (!this->ReopenCatalog(job)) {
355 ✗ atomic_inc32(&num_errors_);
356 ✗ NotifyFinished();
357 ✗ return;
358 }
359 }
360
1/2
✓ Branch 0 taken 17973521 times.
✗ Branch 1 not taken.
 17973521 if (serialize_callbacks_) {
361 17973521 MutexLockGuard const m(&catalog_callback_lock_);
362
2/4
✓ Branch 1 taken 18041680 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 18041680 times.
✗ Branch 5 not taken.
 18041680 this->NotifyListeners(job->GetCallbackData());
363 18041680 } else {
364 ✗ this->NotifyListeners(job->GetCallbackData());
365 }
366
2/2
✓ Branch 0 taken 76222 times.
✓ Branch 1 taken 17965360 times.
 18041582 if (!this->no_close_) {
367
2/4
✓ Branch 0 taken 76222 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 76222 times.
 76222 if (!this->CloseCatalog(true, job)) {
368 ✗ atomic_inc32(&num_errors_);
369 ✗ NotifyFinished();
370 ✗ return;
371 }
372 }
373 18041582 FinalizeJob(job);
374 }
375
376 18041706 void FinalizeJob(CatalogJob *job) {
377 {
378 18041706 MutexLockGuard const m(&catalogs_lock_);
379
1/2
✓ Branch 1 taken 18041902 times.
✗ Branch 2 not taken.
 18041902 catalogs_processing_.Erase(job->hash);
380
1/2
✓ Branch 1 taken 18041902 times.
✗ Branch 2 not taken.
 18041902 catalogs_done_.Insert(job->hash, true);
381 // No more catalogs to process -> finish
382
1/2
✓ Branch 2 taken 3210 times.
✗ Branch 3 not taken.
 18045112 if (catalogs_processing_.size() == 0 && pre_job_queue_.IsEmpty()
383
5/6
✓ Branch 0 taken 3210 times.
✓ Branch 1 taken 18038692 times.
✓ Branch 3 taken 3210 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 3210 times.
✓ Branch 6 taken 18038692 times.
 18045112 && post_job_queue_.IsEmpty()) {
384
1/2
✓ Branch 1 taken 3210 times.
✗ Branch 2 not taken.
 3210 NotifyFinished();
385 }
386 18041902 }
387
2/2
✓ Branch 0 taken 8999619 times.
✓ Branch 1 taken 9041695 times.
 18041314 if (effective_traversal_type_ == Base::kDepthFirst) {
388 8999619 job->WakeParents();
389 }
390
1/2
✓ Branch 0 taken 18040922 times.
✗ Branch 1 not taken.
 18040824 delete job;
391 18038668 }
392
393 8998087 void OnChildFinished(const int &a, CatalogJob *job) {
394 // atomic_xadd32 returns value before subtraction -> needs to equal 1
395
2/2
✓ Branch 1 taken 1664470 times.
✓ Branch 2 taken 7334401 times.
 8998087 if (atomic_xadd32(&job->children_unprocessed, -1) == 1) {
396 1664470 post_job_queue_.EnqueueFront(job);
397 }
398 8998822 }
399
400 unsigned int num_threads_;
401 bool serialize_callbacks_;
402
403 uint64_t effective_history_depth_;
404 time_t effective_timestamp_threshold_;
405 TraversalType effective_traversal_type_;
406
407 pthread_t *threads_process_;
408 atomic_int32 num_errors_;
409
410 Tube<CatalogJob> pre_job_queue_;
411 Tube<CatalogJob> post_job_queue_;
412 SmallHashDynamic<shash::Any, CatalogJob *> catalogs_processing_;
413 SmallHashDynamic<shash::Any, bool> catalogs_done_;
414 pthread_mutex_t catalogs_lock_;
415
416 pthread_mutex_t catalog_callback_lock_;
417 };
418
419 } // namespace swissknife
420
421 #endif // CVMFS_CATALOG_TRAVERSAL_PARALLEL_H_
422