| Directory: | cvmfs/ |
|---|---|
| File: | cvmfs/catalog_traversal_parallel.h |
| Date: | 2025-12-28 02:35:52 |
| 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 | 982 | explicit CatalogTraversalParallel(const Parameters ¶ms) | |
| 43 | : CatalogTraversalBase<ObjectFetcherT>(params) | ||
| 44 | 982 | , num_threads_(params.num_threads) | |
| 45 |
4/8✓ Branch 2 taken 982 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 982 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 982 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 982 times.
✗ Branch 12 not taken.
|
982 | , serialize_callbacks_(params.serialize_callbacks) { |
| 46 | 982 | atomic_init32(&num_errors_); | |
| 47 |
1/2✓ Branch 1 taken 982 times.
✗ Branch 2 not taken.
|
982 | shash::Any null_hash; |
| 48 | 982 | null_hash.SetNull(); | |
| 49 |
1/2✓ Branch 1 taken 982 times.
✗ Branch 2 not taken.
|
982 | catalogs_processing_.Init(1024, null_hash, hasher); |
| 50 |
1/2✓ Branch 1 taken 982 times.
✗ Branch 2 not taken.
|
982 | catalogs_done_.Init(1024, null_hash, hasher); |
| 51 | 982 | pthread_mutex_init(&catalog_callback_lock_, NULL); | |
| 52 | 982 | pthread_mutex_init(&catalogs_lock_, NULL); | |
| 53 | 982 | effective_history_depth_ = this->default_history_depth_; | |
| 54 | 982 | effective_timestamp_threshold_ = this->default_timestamp_threshold_; | |
| 55 | 982 | } | |
| 56 | |||
| 57 | protected: | ||
| 58 | struct CatalogJob : public CatalogTraversal<ObjectFetcherT>::CatalogJob, | ||
| 59 | public Observable<int> { | ||
| 60 | 6631522 | 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 | 6631522 | history_depth, parent) { | |
| 67 | 6631522 | atomic_init32(&children_unprocessed); | |
| 68 | 6631522 | } | |
| 69 | |||
| 70 |
1/2✓ Branch 1 taken 3306068 times.
✗ Branch 2 not taken.
|
3306410 | 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 | 678 | bool Traverse(const TraversalType type = Base::kBreadthFirst) { | |
| 86 |
1/2✓ Branch 1 taken 678 times.
✗ Branch 2 not taken.
|
678 | const shash::Any root_catalog_hash = this->GetRepositoryRootCatalogHash(); |
| 87 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 678 times.
|
678 | if (root_catalog_hash.IsNull()) { |
| 88 | ✗ | return false; | |
| 89 | } | ||
| 90 |
1/2✓ Branch 1 taken 678 times.
✗ Branch 2 not taken.
|
678 | 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 | 948 | 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 | 1896 | if (this->no_repeat_history_ | |
| 104 |
4/6✓ Branch 0 taken 552 times.
✓ Branch 1 taken 396 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 552 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 948 times.
|
948 | && catalogs_done_.Contains(root_catalog_hash)) { |
| 105 | ✗ | return true; | |
| 106 | } | ||
| 107 | 948 | effective_traversal_type_ = type; | |
| 108 |
3/6✓ Branch 2 taken 948 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 948 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 948 times.
✗ Branch 9 not taken.
|
948 | CatalogJob *root_job = new CatalogJob("", root_catalog_hash, 0, 0); |
| 109 | 948 | PushJob(root_job); | |
| 110 | 948 | 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 | 798 | bool TraverseList(const HashList &root_catalog_list, | |
| 119 | const TraversalType type = Base::kBreadthFirst) { | ||
| 120 | // Push in reverse order for CatalogTraversal-like behavior | ||
| 121 | 798 | HashList::const_reverse_iterator i = root_catalog_list.rbegin(); | |
| 122 | 798 | const HashList::const_reverse_iterator iend = root_catalog_list.rend(); | |
| 123 | 798 | bool has_pushed = false; | |
| 124 | { | ||
| 125 | 798 | MutexLockGuard m(&catalogs_lock_); | |
| 126 |
3/4✓ Branch 2 taken 2440 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1642 times.
✓ Branch 5 taken 798 times.
|
2440 | for (; i != iend; ++i) { |
| 127 |
7/8✓ Branch 0 taken 1480 times.
✓ Branch 1 taken 162 times.
✓ Branch 4 taken 1480 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 544 times.
✓ Branch 7 taken 936 times.
✓ Branch 8 taken 544 times.
✓ Branch 9 taken 1098 times.
|
1642 | if (this->no_repeat_history_ && catalogs_done_.Contains(*i)) { |
| 128 | 544 | continue; | |
| 129 | } | ||
| 130 | |||
| 131 |
3/6✓ Branch 2 taken 1098 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 1098 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 1098 times.
✗ Branch 10 not taken.
|
1098 | CatalogJob *root_job = new CatalogJob("", *i, 0, 0); |
| 132 |
1/2✓ Branch 1 taken 1098 times.
✗ Branch 2 not taken.
|
1098 | PushJobUnlocked(root_job); |
| 133 | 1098 | has_pushed = true; | |
| 134 | } | ||
| 135 | 798 | } | |
| 136 | // noop: no catalogs to traverse | ||
| 137 |
2/2✓ Branch 0 taken 208 times.
✓ Branch 1 taken 590 times.
|
798 | if (!has_pushed) { |
| 138 | 208 | return true; | |
| 139 | } | ||
| 140 | 590 | effective_traversal_type_ = type; | |
| 141 | 590 | effective_history_depth_ = Parameters::kNoHistory; | |
| 142 | 590 | effective_timestamp_threshold_ = Parameters::kNoTimestampThreshold; | |
| 143 |
1/2✓ Branch 1 taken 590 times.
✗ Branch 2 not taken.
|
590 | bool result = DoTraverse(); |
| 144 | 590 | effective_history_depth_ = this->default_history_depth_; | |
| 145 | 590 | effective_timestamp_threshold_ = this->default_timestamp_threshold_; | |
| 146 | 590 | 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 | 36 | bool TraverseRevision(const shash::Any &root_catalog_hash, | |
| 158 | const TraversalType type = Base::kBreadthFirst) { | ||
| 159 | 36 | effective_history_depth_ = Parameters::kNoHistory; | |
| 160 | 36 | effective_timestamp_threshold_ = Parameters::kNoTimestampThreshold; | |
| 161 | 36 | bool result = Traverse(root_catalog_hash, type); | |
| 162 | 36 | effective_history_depth_ = this->default_history_depth_; | |
| 163 | 36 | effective_timestamp_threshold_ = this->default_timestamp_threshold_; | |
| 164 | 36 | return result; | |
| 165 | } | ||
| 166 | |||
| 167 | protected: | ||
| 168 | 30500468 | 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 | 30500468 | *(reinterpret_cast<const uint32_t *>(key.digest) + 1)); | |
| 172 | } | ||
| 173 | |||
| 174 | 1538 | 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 | 1538 | threads_process_ = reinterpret_cast<pthread_t *>( | |
| 178 | 1538 | smalloc(sizeof(pthread_t) * num_threads_)); | |
| 179 |
2/2✓ Branch 0 taken 1790 times.
✓ Branch 1 taken 1538 times.
|
3328 | for (unsigned int i = 0; i < num_threads_; ++i) { |
| 180 | 1790 | int retval = pthread_create(&threads_process_[i], NULL, MainProcessQueue, | |
| 181 | this); | ||
| 182 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 1790 times.
|
1790 | if (retval != 0) |
| 183 | ✗ | PANIC(kLogStderr, "failed to create thread"); | |
| 184 | } | ||
| 185 | |||
| 186 |
2/2✓ Branch 0 taken 1790 times.
✓ Branch 1 taken 1538 times.
|
3328 | for (unsigned int i = 0; i < num_threads_; ++i) { |
| 187 | 1790 | int retval = pthread_join(threads_process_[i], NULL); | |
| 188 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 1790 times.
|
1790 | assert(retval == 0); |
| 189 | } | ||
| 190 | 1538 | free(threads_process_); | |
| 191 | |||
| 192 |
2/2✓ Branch 1 taken 18 times.
✓ Branch 2 taken 1520 times.
|
1538 | if (atomic_read32(&num_errors_)) |
| 193 | 18 | return false; | |
| 194 | |||
| 195 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 1520 times.
|
1520 | assert(catalogs_processing_.size() == 0); |
| 196 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 1520 times.
|
1520 | assert(pre_job_queue_.IsEmpty()); |
| 197 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 1520 times.
|
1520 | assert(post_job_queue_.IsEmpty()); |
| 198 | 1520 | return true; | |
| 199 | } | ||
| 200 | |||
| 201 | 1790 | static void *MainProcessQueue(void *data) { | |
| 202 | 1790 | CatalogTraversalParallel<ObjectFetcherT> *traversal = reinterpret_cast< | |
| 203 | CatalogTraversalParallel<ObjectFetcherT> *>(data); | ||
| 204 | CatalogJob *current_job; | ||
| 205 | while (true) { | ||
| 206 | 7241308 | current_job = traversal->post_job_queue_.TryPopFront(); | |
| 207 |
2/2✓ Branch 0 taken 611622 times.
✓ Branch 1 taken 6631072 times.
|
7242694 | if (current_job != NULL) { |
| 208 | 611622 | traversal->ProcessJobPost(current_job); | |
| 209 | } else { | ||
| 210 | 6631072 | current_job = traversal->pre_job_queue_.PopFront(); | |
| 211 | // NULL means the master thread tells us to finish | ||
| 212 |
2/2✓ Branch 1 taken 1790 times.
✓ Branch 2 taken 6629624 times.
|
6631450 | if (current_job->hash.IsNull()) { |
| 213 |
1/2✓ Branch 0 taken 1790 times.
✗ Branch 1 not taken.
|
1790 | delete current_job; |
| 214 | 1772 | break; | |
| 215 | } | ||
| 216 | 6629624 | traversal->ProcessJobPre(current_job); | |
| 217 | } | ||
| 218 | } | ||
| 219 | 1772 | return NULL; | |
| 220 | } | ||
| 221 | |||
| 222 | 1538 | void NotifyFinished() { | |
| 223 |
1/2✓ Branch 1 taken 1538 times.
✗ Branch 2 not taken.
|
1538 | shash::Any null_hash; |
| 224 | 1538 | null_hash.SetNull(); | |
| 225 |
2/2✓ Branch 0 taken 1790 times.
✓ Branch 1 taken 1538 times.
|
3328 | for (unsigned i = 0; i < num_threads_; ++i) { |
| 226 |
3/6✓ Branch 2 taken 1790 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 1790 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 1790 times.
✗ Branch 9 not taken.
|
1790 | CatalogJob *job = new CatalogJob("", null_hash, 0, 0); |
| 227 |
1/2✓ Branch 1 taken 1790 times.
✗ Branch 2 not taken.
|
1790 | pre_job_queue_.EnqueueFront(job); |
| 228 | } | ||
| 229 | 1538 | } | |
| 230 | |||
| 231 | 948 | void PushJob(CatalogJob *job) { | |
| 232 | 948 | MutexLockGuard m(&catalogs_lock_); | |
| 233 |
1/2✓ Branch 1 taken 948 times.
✗ Branch 2 not taken.
|
948 | PushJobUnlocked(job); |
| 234 | 948 | } | |
| 235 | |||
| 236 | 6629732 | void PushJobUnlocked(CatalogJob *job) { | |
| 237 | 6629732 | catalogs_processing_.Insert(job->hash, job); | |
| 238 | 6629732 | pre_job_queue_.EnqueueFront(job); | |
| 239 | 6629732 | } | |
| 240 | |||
| 241 | 6629480 | void ProcessJobPre(CatalogJob *job) { | |
| 242 |
4/6✓ Branch 0 taken 6629498 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 6627878 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 18 times.
✓ Branch 6 taken 6627860 times.
|
6629480 | if (!this->PrepareCatalog(job)) { |
| 243 | 18 | atomic_inc32(&num_errors_); | |
| 244 |
1/2✓ Branch 1 taken 18 times.
✗ Branch 2 not taken.
|
18 | NotifyFinished(); |
| 245 | 6016958 | return; | |
| 246 | } | ||
| 247 |
2/2✓ Branch 0 taken 102 times.
✓ Branch 1 taken 6627758 times.
|
6627860 | if (job->ignore) { |
| 248 |
1/2✓ Branch 1 taken 102 times.
✗ Branch 2 not taken.
|
102 | FinalizeJob(job); |
| 249 | 102 | return; | |
| 250 | } | ||
| 251 |
1/2✓ Branch 1 taken 6612836 times.
✗ Branch 2 not taken.
|
6627758 | 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 | 6612836 | MutexLockGuard m(&catalogs_lock_); | |
| 257 |
2/2✓ Branch 0 taken 3323090 times.
✓ Branch 1 taken 3306504 times.
|
6629594 | if (effective_traversal_type_ == Base::kBreadthFirst) { |
| 258 |
1/2✓ Branch 1 taken 3323090 times.
✗ Branch 2 not taken.
|
3323090 | num_children = PushPreviousRevision(job) |
| 259 |
1/2✓ Branch 1 taken 3323090 times.
✗ Branch 2 not taken.
|
3323090 | + PushNestedCatalogs(job, catalog_list); |
| 260 | } else { | ||
| 261 |
1/2✓ Branch 1 taken 3306504 times.
✗ Branch 2 not taken.
|
3306504 | num_children = PushNestedCatalogs(job, catalog_list) |
| 262 |
1/2✓ Branch 1 taken 3306504 times.
✗ Branch 2 not taken.
|
3306504 | + PushPreviousRevision(job); |
| 263 | 3306504 | atomic_write32(&job->children_unprocessed, num_children); | |
| 264 | } | ||
| 265 |
3/6✓ Branch 0 taken 6629594 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 6629594 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 6629594 times.
|
6629594 | if (!this->CloseCatalog(false, job)) { |
| 266 | ✗ | atomic_inc32(&num_errors_); | |
| 267 | ✗ | NotifyFinished(); | |
| 268 | } | ||
| 269 | 6629594 | } | |
| 270 | |||
| 271 | // breadth-first: can post-process immediately | ||
| 272 | // depth-first: no children -> can post-process immediately | ||
| 273 |
4/4✓ Branch 0 taken 3306378 times.
✓ Branch 1 taken 3323126 times.
✓ Branch 2 taken 2694756 times.
✓ Branch 3 taken 611622 times.
|
6629504 | if (effective_traversal_type_ == Base::kBreadthFirst || num_children == 0) { |
| 274 |
1/2✓ Branch 1 taken 6016064 times.
✗ Branch 2 not taken.
|
6017882 | ProcessJobPost(job); |
| 275 | 6016064 | return; | |
| 276 | } | ||
| 277 |
2/2✓ Branch 1 taken 611622 times.
✓ Branch 2 taken 6016838 times.
|
6627686 | } |
| 278 | |||
| 279 | 6629594 | unsigned int PushNestedCatalogs(CatalogJob *job, | |
| 280 | const NestedCatalogList &catalog_list) { | ||
| 281 | 6629594 | typename NestedCatalogList::const_iterator i = catalog_list.begin(); | |
| 282 | 6629594 | typename NestedCatalogList::const_iterator iend = catalog_list.end(); | |
| 283 | 6629594 | unsigned int num_children = 0; | |
| 284 |
2/2✓ Branch 2 taken 6627678 times.
✓ Branch 3 taken 6629594 times.
|
13257272 | for (; i != iend; ++i) { |
| 285 |
7/8✓ Branch 0 taken 11022 times.
✓ Branch 1 taken 6616656 times.
✓ Branch 4 taken 11022 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 1182 times.
✓ Branch 7 taken 9840 times.
✓ Branch 8 taken 1182 times.
✓ Branch 9 taken 6626496 times.
|
6627678 | if (this->no_repeat_history_ && catalogs_done_.Contains(i->hash)) { |
| 286 | 1182 | continue; | |
| 287 | } | ||
| 288 | |||
| 289 | CatalogJob *child; | ||
| 290 | 13252992 | if (!this->no_repeat_history_ | |
| 291 |
7/8✓ Branch 0 taken 9840 times.
✓ Branch 1 taken 6616656 times.
✓ Branch 4 taken 9840 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 9696 times.
✓ Branch 7 taken 144 times.
✓ Branch 8 taken 6626352 times.
✓ Branch 9 taken 144 times.
|
6626496 | || !catalogs_processing_.Lookup(i->hash, &child)) { |
| 292 |
2/2✓ Branch 0 taken 6599466 times.
✓ Branch 1 taken 26886 times.
|
6626352 | CatalogTN *parent = (this->no_close_) ? job->catalog : NULL; |
| 293 |
1/2✓ Branch 2 taken 6626352 times.
✗ Branch 3 not taken.
|
13252704 | child = new CatalogJob(i->mountpoint.ToString(), |
| 294 |
1/2✓ Branch 2 taken 6626352 times.
✗ Branch 3 not taken.
|
6626352 | i->hash, |
| 295 | 6626352 | job->tree_level + 1, | |
| 296 |
1/2✓ Branch 1 taken 6626352 times.
✗ Branch 2 not taken.
|
6626352 | job->history_depth, |
| 297 | parent); | ||
| 298 |
1/2✓ Branch 1 taken 6626352 times.
✗ Branch 2 not taken.
|
6626352 | PushJobUnlocked(child); |
| 299 | } | ||
| 300 | |||
| 301 |
2/2✓ Branch 0 taken 3305796 times.
✓ Branch 1 taken 3320700 times.
|
6626496 | if (effective_traversal_type_ == Base::kDepthFirst) { |
| 302 |
1/2✓ Branch 1 taken 3305796 times.
✗ Branch 2 not taken.
|
3305796 | child->RegisterListener(&CatalogTraversalParallel::OnChildFinished, |
| 303 | this, job); | ||
| 304 | } | ||
| 305 | 6626496 | ++num_children; | |
| 306 | } | ||
| 307 | 6629594 | 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 | 6629594 | unsigned int PushPreviousRevision(CatalogJob *job) { | |
| 315 | // only root catalogs are used for entering a previous revision (graph) | ||
| 316 |
2/2✓ Branch 1 taken 6626300 times.
✓ Branch 2 taken 3294 times.
|
6629594 | if (!job->catalog->IsRoot()) { |
| 317 | 6626300 | return 0; | |
| 318 | } | ||
| 319 | |||
| 320 |
1/2✓ Branch 1 taken 3294 times.
✗ Branch 2 not taken.
|
3294 | const shash::Any previous_revision = job->catalog->GetPreviousRevision(); |
| 321 |
2/2✓ Branch 1 taken 318 times.
✓ Branch 2 taken 2976 times.
|
3294 | if (previous_revision.IsNull()) { |
| 322 | 318 | return 0; | |
| 323 | } | ||
| 324 | |||
| 325 | // check if the next deeper history level is actually requested | ||
| 326 |
3/4✓ Branch 1 taken 2976 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 1642 times.
✓ Branch 4 taken 1334 times.
|
2976 | if (this->IsBelowPruningThresholds(*job, effective_history_depth_, |
| 327 | effective_timestamp_threshold_)) { | ||
| 328 | 1642 | return 0; | |
| 329 | } | ||
| 330 | |||
| 331 | 2668 | if (this->no_repeat_history_ | |
| 332 |
5/8✓ Branch 0 taken 830 times.
✓ Branch 1 taken 504 times.
✓ Branch 3 taken 830 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 830 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 1334 times.
|
1334 | && catalogs_done_.Contains(previous_revision)) { |
| 333 | ✗ | return 0; | |
| 334 | } | ||
| 335 | |||
| 336 | CatalogJob *prev_job; | ||
| 337 | 2668 | if (!this->no_repeat_history_ | |
| 338 |
5/8✓ Branch 0 taken 830 times.
✓ Branch 1 taken 504 times.
✓ Branch 3 taken 830 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 830 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 1334 times.
✗ Branch 8 not taken.
|
1334 | || !catalogs_processing_.Lookup(previous_revision, &prev_job)) { |
| 339 |
2/4✓ Branch 2 taken 1334 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 1334 times.
✗ Branch 6 not taken.
|
2668 | prev_job = new CatalogJob("", previous_revision, 0, |
| 340 |
1/2✓ Branch 1 taken 1334 times.
✗ Branch 2 not taken.
|
1334 | job->history_depth + 1); |
| 341 |
1/2✓ Branch 1 taken 1334 times.
✗ Branch 2 not taken.
|
1334 | PushJobUnlocked(prev_job); |
| 342 | } | ||
| 343 | |||
| 344 |
2/2✓ Branch 0 taken 270 times.
✓ Branch 1 taken 1064 times.
|
1334 | if (effective_traversal_type_ == Base::kDepthFirst) { |
| 345 |
1/2✓ Branch 1 taken 270 times.
✗ Branch 2 not taken.
|
270 | prev_job->RegisterListener(&CatalogTraversalParallel::OnChildFinished, |
| 346 | this, job); | ||
| 347 | } | ||
| 348 | 1334 | return 1; | |
| 349 | } | ||
| 350 | |||
| 351 | 6628352 | void ProcessJobPost(CatalogJob *job) { | |
| 352 | // Save time by keeping catalog open when suitable | ||
| 353 |
1/2✓ Branch 0 taken 6628352 times.
✗ Branch 1 not taken.
|
6628352 | if (job->catalog == NULL) { |
| 354 |
2/4✓ Branch 0 taken 6628352 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 6613196 times.
|
6628352 | if (!this->ReopenCatalog(job)) { |
| 355 | ✗ | atomic_inc32(&num_errors_); | |
| 356 | ✗ | NotifyFinished(); | |
| 357 | ✗ | return; | |
| 358 | } | ||
| 359 | } | ||
| 360 |
1/2✓ Branch 0 taken 6613196 times.
✗ Branch 1 not taken.
|
6613196 | if (serialize_callbacks_) { |
| 361 | 6613196 | MutexLockGuard m(&catalog_callback_lock_); | |
| 362 |
2/4✓ Branch 1 taken 6629594 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 6629594 times.
✗ Branch 5 not taken.
|
6629594 | this->NotifyListeners(job->GetCallbackData()); |
| 363 | 6629594 | } else { | |
| 364 | ✗ | this->NotifyListeners(job->GetCallbackData()); | |
| 365 | } | ||
| 366 |
2/2✓ Branch 0 taken 30038 times.
✓ Branch 1 taken 6599538 times.
|
6629576 | if (!this->no_close_) { |
| 367 |
2/4✓ Branch 0 taken 30038 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 30038 times.
|
30038 | if (!this->CloseCatalog(true, job)) { |
| 368 | ✗ | atomic_inc32(&num_errors_); | |
| 369 | ✗ | NotifyFinished(); | |
| 370 | ✗ | return; | |
| 371 | } | ||
| 372 | } | ||
| 373 | 6629576 | FinalizeJob(job); | |
| 374 | } | ||
| 375 | |||
| 376 | 6629660 | void FinalizeJob(CatalogJob *job) { | |
| 377 | { | ||
| 378 | 6629660 | MutexLockGuard m(&catalogs_lock_); | |
| 379 |
1/2✓ Branch 1 taken 6629696 times.
✗ Branch 2 not taken.
|
6629696 | catalogs_processing_.Erase(job->hash); |
| 380 |
1/2✓ Branch 1 taken 6629696 times.
✗ Branch 2 not taken.
|
6629696 | catalogs_done_.Insert(job->hash, true); |
| 381 | // No more catalogs to process -> finish | ||
| 382 |
1/2✓ Branch 2 taken 1520 times.
✗ Branch 3 not taken.
|
6631216 | if (catalogs_processing_.size() == 0 && pre_job_queue_.IsEmpty() |
| 383 |
5/6✓ Branch 0 taken 1520 times.
✓ Branch 1 taken 6628176 times.
✓ Branch 3 taken 1520 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 1520 times.
✓ Branch 6 taken 6628176 times.
|
6631216 | && post_job_queue_.IsEmpty()) { |
| 384 |
1/2✓ Branch 1 taken 1520 times.
✗ Branch 2 not taken.
|
1520 | NotifyFinished(); |
| 385 | } | ||
| 386 | 6629696 | } | |
| 387 |
2/2✓ Branch 0 taken 3306428 times.
✓ Branch 1 taken 3323232 times.
|
6629660 | if (effective_traversal_type_ == Base::kDepthFirst) { |
| 388 | 3306428 | job->WakeParents(); | |
| 389 | } | ||
| 390 |
2/2✓ Branch 0 taken 6629120 times.
✓ Branch 1 taken 126 times.
|
6629246 | delete job; |
| 391 | 6627716 | } | |
| 392 | |||
| 393 | 3305850 | void OnChildFinished(const int &a, CatalogJob *job) { | |
| 394 | // atomic_xadd32 returns value before subtraction -> needs to equal 1 | ||
| 395 |
2/2✓ Branch 1 taken 611622 times.
✓ Branch 2 taken 2694354 times.
|
3305850 | if (atomic_xadd32(&job->children_unprocessed, -1) == 1) { |
| 396 | 611622 | post_job_queue_.EnqueueFront(job); | |
| 397 | } | ||
| 398 | 3305976 | } | |
| 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 |