| Directory: | cvmfs/ |
|---|---|
| File: | cvmfs/catalog_traversal_parallel.h |
| Date: | 2026-02-08 02:36:20 |
| Exec | Total | Coverage | |
|---|---|---|---|
| Lines: | 186 | 199 | 93.5% |
| Branches: | 159 | 244 | 65.2% |
| 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 | 2617 | explicit CatalogTraversalParallel(const Parameters ¶ms) | |
| 43 | : CatalogTraversalBase<ObjectFetcherT>(params) | ||
| 44 | 2617 | , num_threads_(params.num_threads) | |
| 45 |
4/8✓ Branch 2 taken 2617 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2617 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2617 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 2617 times.
✗ Branch 12 not taken.
|
2617 | , serialize_callbacks_(params.serialize_callbacks) { |
| 46 | 2617 | atomic_init32(&num_errors_); | |
| 47 |
1/2✓ Branch 1 taken 2617 times.
✗ Branch 2 not taken.
|
2617 | shash::Any null_hash; |
| 48 | 2617 | null_hash.SetNull(); | |
| 49 |
1/2✓ Branch 1 taken 2617 times.
✗ Branch 2 not taken.
|
2617 | catalogs_processing_.Init(1024, null_hash, hasher); |
| 50 |
1/2✓ Branch 1 taken 2617 times.
✗ Branch 2 not taken.
|
2617 | catalogs_done_.Init(1024, null_hash, hasher); |
| 51 | 2617 | pthread_mutex_init(&catalog_callback_lock_, NULL); | |
| 52 | 2617 | pthread_mutex_init(&catalogs_lock_, NULL); | |
| 53 | 2617 | effective_history_depth_ = this->default_history_depth_; | |
| 54 | 2617 | effective_timestamp_threshold_ = this->default_timestamp_threshold_; | |
| 55 | 2617 | } | |
| 56 | |||
| 57 | protected: | ||
| 58 | struct CatalogJob : public CatalogTraversal<ObjectFetcherT>::CatalogJob, | ||
| 59 | public Observable<int> { | ||
| 60 | 18051577 | 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 | 18051577 | history_depth, parent) { | |
| 67 | 18051577 | atomic_init32(&children_unprocessed); | |
| 68 | 18051577 | } | |
| 69 | |||
| 70 |
1/2✓ Branch 1 taken 9000021 times.
✗ Branch 2 not taken.
|
9000854 | 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 | 1792 | bool Traverse(const TraversalType type = Base::kBreadthFirst) { | |
| 86 |
1/2✓ Branch 1 taken 1792 times.
✗ Branch 2 not taken.
|
1792 | const shash::Any root_catalog_hash = this->GetRepositoryRootCatalogHash(); |
| 87 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 1792 times.
|
1792 | if (root_catalog_hash.IsNull()) { |
| 88 | ✗ | return false; | |
| 89 | } | ||
| 90 |
1/2✓ Branch 1 taken 1792 times.
✗ Branch 2 not taken.
|
1792 | 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 | 2527 | 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 | 5054 | if (this->no_repeat_history_ | |
| 104 |
4/6✓ Branch 0 taken 1449 times.
✓ Branch 1 taken 1078 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1449 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2527 times.
|
2527 | && catalogs_done_.Contains(root_catalog_hash)) { |
| 105 | ✗ | return true; | |
| 106 | } | ||
| 107 | 2527 | effective_traversal_type_ = type; | |
| 108 |
3/6✓ Branch 2 taken 2527 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2527 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2527 times.
✗ Branch 9 not taken.
|
2527 | CatalogJob *root_job = new CatalogJob("", root_catalog_hash, 0, 0); |
| 109 | 2527 | PushJob(root_job); | |
| 110 | 2527 | 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 | 2065 | bool TraverseList(const HashList &root_catalog_list, | |
| 119 | const TraversalType type = Base::kBreadthFirst) { | ||
| 120 | // Push in reverse order for CatalogTraversal-like behavior | ||
| 121 | 2065 | HashList::const_reverse_iterator i = root_catalog_list.rbegin(); | |
| 122 | 2065 | const HashList::const_reverse_iterator iend = root_catalog_list.rend(); | |
| 123 | 2065 | bool has_pushed = false; | |
| 124 | { | ||
| 125 | 2065 | MutexLockGuard m(&catalogs_lock_); | |
| 126 |
3/4✓ Branch 2 taken 6333 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 4268 times.
✓ Branch 5 taken 2065 times.
|
6333 | for (; i != iend; ++i) { |
| 127 |
7/8✓ Branch 0 taken 3827 times.
✓ Branch 1 taken 441 times.
✓ Branch 4 taken 3827 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 1394 times.
✓ Branch 7 taken 2433 times.
✓ Branch 8 taken 1394 times.
✓ Branch 9 taken 2874 times.
|
4268 | if (this->no_repeat_history_ && catalogs_done_.Contains(*i)) { |
| 128 | 1394 | continue; | |
| 129 | } | ||
| 130 | |||
| 131 |
3/6✓ Branch 2 taken 2874 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 2874 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 2874 times.
✗ Branch 10 not taken.
|
2874 | CatalogJob *root_job = new CatalogJob("", *i, 0, 0); |
| 132 |
1/2✓ Branch 1 taken 2874 times.
✗ Branch 2 not taken.
|
2874 | PushJobUnlocked(root_job); |
| 133 | 2874 | has_pushed = true; | |
| 134 | } | ||
| 135 | 2065 | } | |
| 136 | // noop: no catalogs to traverse | ||
| 137 |
2/2✓ Branch 0 taken 533 times.
✓ Branch 1 taken 1532 times.
|
2065 | if (!has_pushed) { |
| 138 | 533 | return true; | |
| 139 | } | ||
| 140 | 1532 | effective_traversal_type_ = type; | |
| 141 | 1532 | effective_history_depth_ = Parameters::kNoHistory; | |
| 142 | 1532 | effective_timestamp_threshold_ = Parameters::kNoTimestampThreshold; | |
| 143 |
1/2✓ Branch 1 taken 1532 times.
✗ Branch 2 not taken.
|
1532 | bool result = DoTraverse(); |
| 144 | 1532 | effective_history_depth_ = this->default_history_depth_; | |
| 145 | 1532 | effective_timestamp_threshold_ = this->default_timestamp_threshold_; | |
| 146 | 1532 | 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 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 | 82533325 | 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 | 82533325 | *(reinterpret_cast<const uint32_t *>(key.digest) + 1)); | |
| 172 | } | ||
| 173 | |||
| 174 | 4059 | 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 | 4059 | threads_process_ = reinterpret_cast<pthread_t *>( | |
| 178 | 4059 | smalloc(sizeof(pthread_t) * num_threads_)); | |
| 179 |
2/2✓ Branch 0 taken 4745 times.
✓ Branch 1 taken 4059 times.
|
8804 | for (unsigned int i = 0; i < num_threads_; ++i) { |
| 180 | 4745 | int retval = pthread_create(&threads_process_[i], NULL, MainProcessQueue, | |
| 181 | this); | ||
| 182 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 4745 times.
|
4745 | if (retval != 0) |
| 183 | ✗ | PANIC(kLogStderr, "failed to create thread"); | |
| 184 | } | ||
| 185 | |||
| 186 |
2/2✓ Branch 0 taken 4745 times.
✓ Branch 1 taken 4059 times.
|
8804 | for (unsigned int i = 0; i < num_threads_; ++i) { |
| 187 | 4745 | int retval = pthread_join(threads_process_[i], NULL); | |
| 188 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 4745 times.
|
4745 | assert(retval == 0); |
| 189 | } | ||
| 190 | 4059 | free(threads_process_); | |
| 191 | |||
| 192 |
2/2✓ Branch 1 taken 49 times.
✓ Branch 2 taken 4010 times.
|
4059 | if (atomic_read32(&num_errors_)) |
| 193 | 49 | return false; | |
| 194 | |||
| 195 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 4010 times.
|
4010 | assert(catalogs_processing_.size() == 0); |
| 196 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 4010 times.
|
4010 | assert(pre_job_queue_.IsEmpty()); |
| 197 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 4010 times.
|
4010 | assert(post_job_queue_.IsEmpty()); |
| 198 | 4010 | return true; | |
| 199 | } | ||
| 200 | |||
| 201 | 4745 | static void *MainProcessQueue(void *data) { | |
| 202 | 4745 | CatalogTraversalParallel<ObjectFetcherT> *traversal = reinterpret_cast< | |
| 203 | CatalogTraversalParallel<ObjectFetcherT> *>(data); | ||
| 204 | CatalogJob *current_job; | ||
| 205 | while (true) { | ||
| 206 | 19711628 | current_job = traversal->post_job_queue_.TryPopFront(); | |
| 207 |
2/2✓ Branch 0 taken 1664902 times.
✓ Branch 1 taken 18050744 times.
|
19715646 | if (current_job != NULL) { |
| 208 | 1664902 | traversal->ProcessJobPost(current_job); | |
| 209 | } else { | ||
| 210 | 18050744 | current_job = traversal->pre_job_queue_.PopFront(); | |
| 211 | // NULL means the master thread tells us to finish | ||
| 212 |
2/2✓ Branch 1 taken 4745 times.
✓ Branch 2 taken 18046685 times.
|
18051381 | if (current_job->hash.IsNull()) { |
| 213 |
1/2✓ Branch 0 taken 4745 times.
✗ Branch 1 not taken.
|
4745 | delete current_job; |
| 214 | 4745 | break; | |
| 215 | } | ||
| 216 | 18046685 | traversal->ProcessJobPre(current_job); | |
| 217 | } | ||
| 218 | } | ||
| 219 | 4745 | return NULL; | |
| 220 | } | ||
| 221 | |||
| 222 | 4059 | void NotifyFinished() { | |
| 223 |
1/2✓ Branch 1 taken 4059 times.
✗ Branch 2 not taken.
|
4059 | shash::Any null_hash; |
| 224 | 4059 | null_hash.SetNull(); | |
| 225 |
2/2✓ Branch 0 taken 4745 times.
✓ Branch 1 taken 4059 times.
|
8804 | for (unsigned i = 0; i < num_threads_; ++i) { |
| 226 |
3/6✓ Branch 2 taken 4745 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 4745 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 4745 times.
✗ Branch 9 not taken.
|
4745 | CatalogJob *job = new CatalogJob("", null_hash, 0, 0); |
| 227 |
1/2✓ Branch 1 taken 4745 times.
✗ Branch 2 not taken.
|
4745 | pre_job_queue_.EnqueueFront(job); |
| 228 | } | ||
| 229 | 4059 | } | |
| 230 | |||
| 231 | 2527 | void PushJob(CatalogJob *job) { | |
| 232 | 2527 | MutexLockGuard m(&catalogs_lock_); | |
| 233 |
1/2✓ Branch 1 taken 2527 times.
✗ Branch 2 not taken.
|
2527 | PushJobUnlocked(job); |
| 234 | 2527 | } | |
| 235 | |||
| 236 | 18046832 | void PushJobUnlocked(CatalogJob *job) { | |
| 237 | 18046832 | catalogs_processing_.Insert(job->hash, job); | |
| 238 | 18046832 | pre_job_queue_.EnqueueFront(job); | |
| 239 | 18046832 | } | |
| 240 | |||
| 241 | 18046391 | void ProcessJobPre(CatalogJob *job) { | |
| 242 |
4/6✓ Branch 0 taken 18046391 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 18044676 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 49 times.
✓ Branch 6 taken 18044627 times.
|
18046391 | 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 | 16378010 | return; | |
| 246 | } | ||
| 247 |
2/2✓ Branch 0 taken 270 times.
✓ Branch 1 taken 18044357 times.
|
18044627 | if (job->ignore) { |
| 248 |
1/2✓ Branch 1 taken 270 times.
✗ Branch 2 not taken.
|
270 | FinalizeJob(job); |
| 249 | 270 | return; | |
| 250 | } | ||
| 251 |
1/2✓ Branch 1 taken 18007705 times.
✗ Branch 2 not taken.
|
18044357 | NestedCatalogList 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 | 18007705 | MutexLockGuard m(&catalogs_lock_); | |
| 257 |
2/2✓ Branch 0 taken 9045594 times.
✓ Branch 1 taken 9000870 times.
|
18046464 | if (effective_traversal_type_ == Base::kBreadthFirst) { |
| 258 |
1/2✓ Branch 1 taken 9045594 times.
✗ Branch 2 not taken.
|
9045594 | num_children = PushPreviousRevision(job) |
| 259 |
1/2✓ Branch 1 taken 9045594 times.
✗ Branch 2 not taken.
|
9045594 | + PushNestedCatalogs(job, catalog_list); |
| 260 | } else { | ||
| 261 |
1/2✓ Branch 1 taken 9000870 times.
✗ Branch 2 not taken.
|
9000870 | num_children = PushNestedCatalogs(job, catalog_list) |
| 262 |
1/2✓ Branch 1 taken 9000870 times.
✗ Branch 2 not taken.
|
9000870 | + PushPreviousRevision(job); |
| 263 | 9000870 | atomic_write32(&job->children_unprocessed, num_children); | |
| 264 | } | ||
| 265 |
3/6✓ Branch 0 taken 18046464 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 18046464 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 18046464 times.
|
18046464 | if (!this->CloseCatalog(false, job)) { |
| 266 | ✗ | atomic_inc32(&num_errors_); | |
| 267 | ✗ | NotifyFinished(); | |
| 268 | } | ||
| 269 | 18046464 | } | |
| 270 | |||
| 271 | // breadth-first: can post-process immediately | ||
| 272 | // depth-first: no children -> can post-process immediately | ||
| 273 |
4/4✓ Branch 0 taken 9000527 times.
✓ Branch 1 taken 9045839 times.
✓ Branch 2 taken 7335625 times.
✓ Branch 3 taken 1664902 times.
|
18046366 | if (effective_traversal_type_ == Base::kBreadthFirst || num_children == 0) { |
| 274 |
1/2✓ Branch 1 taken 16377152 times.
✗ Branch 2 not taken.
|
16381464 | ProcessJobPost(job); |
| 275 | 16377152 | return; | |
| 276 | } | ||
| 277 |
2/2✓ Branch 1 taken 1664902 times.
✓ Branch 2 taken 16377691 times.
|
18042054 | } |
| 278 | |||
| 279 | 18046464 | unsigned int PushNestedCatalogs(CatalogJob *job, | |
| 280 | const NestedCatalogList &catalog_list) { | ||
| 281 | 18046464 | typename NestedCatalogList::const_iterator i = catalog_list.begin(); | |
| 282 | 18046464 | typename NestedCatalogList::const_iterator iend = catalog_list.end(); | |
| 283 | 18046464 | unsigned int num_children = 0; | |
| 284 |
2/2✓ Branch 2 taken 18041445 times.
✓ Branch 3 taken 18046464 times.
|
36087909 | for (; i != iend; ++i) { |
| 285 |
7/8✓ Branch 0 taken 29437 times.
✓ Branch 1 taken 18012008 times.
✓ Branch 4 taken 29437 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 3187 times.
✓ Branch 7 taken 26250 times.
✓ Branch 8 taken 3187 times.
✓ Branch 9 taken 18038258 times.
|
18041445 | if (this->no_repeat_history_ && catalogs_done_.Contains(i->hash)) { |
| 286 | 3187 | continue; | |
| 287 | } | ||
| 288 | |||
| 289 | CatalogJob *child; | ||
| 290 | 36076516 | if (!this->no_repeat_history_ | |
| 291 |
7/8✓ Branch 0 taken 26250 times.
✓ Branch 1 taken 18012008 times.
✓ Branch 4 taken 26250 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 25858 times.
✓ Branch 7 taken 392 times.
✓ Branch 8 taken 18037866 times.
✓ Branch 9 taken 392 times.
|
18038258 | || !catalogs_processing_.Lookup(i->hash, &child)) { |
| 292 |
2/2✓ Branch 0 taken 17965213 times.
✓ Branch 1 taken 72653 times.
|
18037866 | CatalogTN *parent = (this->no_close_) ? job->catalog : NULL; |
| 293 |
1/2✓ Branch 2 taken 18037866 times.
✗ Branch 3 not taken.
|
36075732 | child = new CatalogJob(i->mountpoint.ToString(), |
| 294 |
1/2✓ Branch 2 taken 18037866 times.
✗ Branch 3 not taken.
|
18037866 | i->hash, |
| 295 | 18037866 | job->tree_level + 1, | |
| 296 |
1/2✓ Branch 1 taken 18037866 times.
✗ Branch 2 not taken.
|
18037866 | job->history_depth, |
| 297 | parent); | ||
| 298 |
1/2✓ Branch 1 taken 18037866 times.
✗ Branch 2 not taken.
|
18037866 | PushJobUnlocked(child); |
| 299 | } | ||
| 300 | |||
| 301 |
2/2✓ Branch 0 taken 8999004 times.
✓ Branch 1 taken 9039254 times.
|
18038258 | if (effective_traversal_type_ == Base::kDepthFirst) { |
| 302 |
1/2✓ Branch 1 taken 8999004 times.
✗ Branch 2 not taken.
|
8999004 | child->RegisterListener(&CatalogTraversalParallel::OnChildFinished, |
| 303 | this, job); | ||
| 304 | } | ||
| 305 | 18038258 | ++num_children; | |
| 306 | } | ||
| 307 | 18046464 | 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 | 18046464 | unsigned int PushPreviousRevision(CatalogJob *job) { | |
| 315 | // only root catalogs are used for entering a previous revision (graph) | ||
| 316 |
2/2✓ Branch 1 taken 18037727 times.
✓ Branch 2 taken 8737 times.
|
18046464 | if (!job->catalog->IsRoot()) { |
| 317 | 18037727 | return 0; | |
| 318 | } | ||
| 319 | |||
| 320 |
1/2✓ Branch 1 taken 8737 times.
✗ Branch 2 not taken.
|
8737 | const shash::Any previous_revision = job->catalog->GetPreviousRevision(); |
| 321 |
2/2✓ Branch 1 taken 835 times.
✓ Branch 2 taken 7902 times.
|
8737 | if (previous_revision.IsNull()) { |
| 322 | 835 | return 0; | |
| 323 | } | ||
| 324 | |||
| 325 | // check if the next deeper history level is actually requested | ||
| 326 |
3/4✓ Branch 1 taken 7902 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 4337 times.
✓ Branch 4 taken 3565 times.
|
7902 | if (this->IsBelowPruningThresholds(*job, effective_history_depth_, |
| 327 | effective_timestamp_threshold_)) { | ||
| 328 | 4337 | return 0; | |
| 329 | } | ||
| 330 | |||
| 331 | 7130 | if (this->no_repeat_history_ | |
| 332 |
5/8✓ Branch 0 taken 2193 times.
✓ Branch 1 taken 1372 times.
✓ Branch 3 taken 2193 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 2193 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 3565 times.
|
3565 | && catalogs_done_.Contains(previous_revision)) { |
| 333 | ✗ | return 0; | |
| 334 | } | ||
| 335 | |||
| 336 | CatalogJob *prev_job; | ||
| 337 | 7130 | if (!this->no_repeat_history_ | |
| 338 |
5/8✓ Branch 0 taken 2193 times.
✓ Branch 1 taken 1372 times.
✓ Branch 3 taken 2193 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 2193 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 3565 times.
✗ Branch 8 not taken.
|
3565 | || !catalogs_processing_.Lookup(previous_revision, &prev_job)) { |
| 339 |
2/4✓ Branch 2 taken 3565 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 3565 times.
✗ Branch 6 not taken.
|
7130 | prev_job = new CatalogJob("", previous_revision, 0, |
| 340 |
1/2✓ Branch 1 taken 3565 times.
✗ Branch 2 not taken.
|
3565 | job->history_depth + 1); |
| 341 |
1/2✓ Branch 1 taken 3565 times.
✗ Branch 2 not taken.
|
3565 | PushJobUnlocked(prev_job); |
| 342 | } | ||
| 343 | |||
| 344 |
2/2✓ Branch 0 taken 735 times.
✓ Branch 1 taken 2830 times.
|
3565 | 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 | 3565 | return 1; | |
| 349 | } | ||
| 350 | |||
| 351 | 18043720 | void ProcessJobPost(CatalogJob *job) { | |
| 352 | // Save time by keeping catalog open when suitable | ||
| 353 |
1/2✓ Branch 0 taken 18043720 times.
✗ Branch 1 not taken.
|
18043720 | if (job->catalog == NULL) { |
| 354 |
2/4✓ Branch 0 taken 18043720 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 18006186 times.
|
18043720 | if (!this->ReopenCatalog(job)) { |
| 355 | ✗ | atomic_inc32(&num_errors_); | |
| 356 | ✗ | NotifyFinished(); | |
| 357 | ✗ | return; | |
| 358 | } | ||
| 359 | } | ||
| 360 |
1/2✓ Branch 0 taken 18006186 times.
✗ Branch 1 not taken.
|
18006186 | if (serialize_callbacks_) { |
| 361 | 18006186 | MutexLockGuard m(&catalog_callback_lock_); | |
| 362 |
2/4✓ Branch 1 taken 18046464 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 18046464 times.
✗ Branch 5 not taken.
|
18046464 | this->NotifyListeners(job->GetCallbackData()); |
| 363 | 18046464 | } else { | |
| 364 | ✗ | this->NotifyListeners(job->GetCallbackData()); | |
| 365 | } | ||
| 366 |
2/2✓ Branch 0 taken 81006 times.
✓ Branch 1 taken 17965311 times.
|
18046317 | if (!this->no_close_) { |
| 367 |
2/4✓ Branch 0 taken 81006 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 81006 times.
|
81006 | if (!this->CloseCatalog(true, job)) { |
| 368 | ✗ | atomic_inc32(&num_errors_); | |
| 369 | ✗ | NotifyFinished(); | |
| 370 | ✗ | return; | |
| 371 | } | ||
| 372 | } | ||
| 373 | 18046317 | FinalizeJob(job); | |
| 374 | } | ||
| 375 | |||
| 376 | 18046391 | void FinalizeJob(CatalogJob *job) { | |
| 377 | { | ||
| 378 | 18046391 | MutexLockGuard m(&catalogs_lock_); | |
| 379 |
1/2✓ Branch 1 taken 18046734 times.
✗ Branch 2 not taken.
|
18046734 | catalogs_processing_.Erase(job->hash); |
| 380 |
1/2✓ Branch 1 taken 18046734 times.
✗ Branch 2 not taken.
|
18046734 | catalogs_done_.Insert(job->hash, true); |
| 381 | // No more catalogs to process -> finish | ||
| 382 |
1/2✓ Branch 2 taken 4010 times.
✗ Branch 3 not taken.
|
18050744 | if (catalogs_processing_.size() == 0 && pre_job_queue_.IsEmpty() |
| 383 |
5/6✓ Branch 0 taken 4010 times.
✓ Branch 1 taken 18042724 times.
✓ Branch 3 taken 4010 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 4010 times.
✓ Branch 6 taken 18042724 times.
|
18050744 | && post_job_queue_.IsEmpty()) { |
| 384 |
1/2✓ Branch 1 taken 4010 times.
✗ Branch 2 not taken.
|
4010 | NotifyFinished(); |
| 385 | } | ||
| 386 | 18046734 | } | |
| 387 |
2/2✓ Branch 0 taken 9000952 times.
✓ Branch 1 taken 9045684 times.
|
18046636 | if (effective_traversal_type_ == Base::kDepthFirst) { |
| 388 | 9000952 | job->WakeParents(); | |
| 389 | } | ||
| 390 |
2/2✓ Branch 0 taken 18045264 times.
✓ Branch 1 taken 294 times.
|
18045558 | delete job; |
| 391 | 18043402 | } | |
| 392 | |||
| 393 | 8999053 | void OnChildFinished(const int &a, CatalogJob *job) { | |
| 394 | // atomic_xadd32 returns value before subtraction -> needs to equal 1 | ||
| 395 |
2/2✓ Branch 1 taken 1664902 times.
✓ Branch 2 taken 7334788 times.
|
8999053 | if (atomic_xadd32(&job->children_unprocessed, -1) == 1) { |
| 396 | 1664902 | post_job_queue_.EnqueueFront(job); | |
| 397 | } | ||
| 398 | 8999690 | } | |
| 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 |