| Directory: | cvmfs/ |
|---|---|
| File: | cvmfs/catalog_traversal_parallel.h |
| Date: | 2025-11-16 02:35:16 |
| 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 | 2574 | explicit CatalogTraversalParallel(const Parameters ¶ms) | |
| 43 | : CatalogTraversalBase<ObjectFetcherT>(params) | ||
| 44 | 2574 | , num_threads_(params.num_threads) | |
| 45 |
4/8✓ Branch 2 taken 2574 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2574 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2574 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 2574 times.
✗ Branch 12 not taken.
|
2574 | , serialize_callbacks_(params.serialize_callbacks) { |
| 46 | 2574 | atomic_init32(&num_errors_); | |
| 47 |
1/2✓ Branch 1 taken 2574 times.
✗ Branch 2 not taken.
|
2574 | shash::Any null_hash; |
| 48 | 2574 | null_hash.SetNull(); | |
| 49 |
1/2✓ Branch 1 taken 2574 times.
✗ Branch 2 not taken.
|
2574 | catalogs_processing_.Init(1024, null_hash, hasher); |
| 50 |
1/2✓ Branch 1 taken 2574 times.
✗ Branch 2 not taken.
|
2574 | catalogs_done_.Init(1024, null_hash, hasher); |
| 51 | 2574 | pthread_mutex_init(&catalog_callback_lock_, NULL); | |
| 52 | 2574 | pthread_mutex_init(&catalogs_lock_, NULL); | |
| 53 | 2574 | effective_history_depth_ = this->default_history_depth_; | |
| 54 | 2574 | effective_timestamp_threshold_ = this->default_timestamp_threshold_; | |
| 55 | 2574 | } | |
| 56 | |||
| 57 | protected: | ||
| 58 | struct CatalogJob : public CatalogTraversal<ObjectFetcherT>::CatalogJob, | ||
| 59 | public Observable<int> { | ||
| 60 | 16213164 | 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 | 16213164 | history_depth, parent) { | |
| 67 | 16213164 | atomic_init32(&children_unprocessed); | |
| 68 | 16213164 | } | |
| 69 | |||
| 70 |
1/2✓ Branch 1 taken 8082564 times.
✗ Branch 2 not taken.
|
8083136 | 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 | 1823 | bool Traverse(const TraversalType type = Base::kBreadthFirst) { | |
| 86 |
1/2✓ Branch 1 taken 1823 times.
✗ Branch 2 not taken.
|
1823 | const shash::Any root_catalog_hash = this->GetRepositoryRootCatalogHash(); |
| 87 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 1823 times.
|
1823 | if (root_catalog_hash.IsNull()) { |
| 88 | ✗ | return false; | |
| 89 | } | ||
| 90 |
1/2✓ Branch 1 taken 1823 times.
✗ Branch 2 not taken.
|
1823 | 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 | 2483 | 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 | 4966 | if (this->no_repeat_history_ | |
| 104 |
4/6✓ Branch 0 taken 1515 times.
✓ Branch 1 taken 968 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1515 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2483 times.
|
2483 | && catalogs_done_.Contains(root_catalog_hash)) { |
| 105 | ✗ | return true; | |
| 106 | } | ||
| 107 | 2483 | effective_traversal_type_ = type; | |
| 108 |
3/6✓ Branch 2 taken 2483 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2483 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2483 times.
✗ Branch 9 not taken.
|
2483 | CatalogJob *root_job = new CatalogJob("", root_catalog_hash, 0, 0); |
| 109 | 2483 | PushJob(root_job); | |
| 110 | 2483 | 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 | 2282 | bool TraverseList(const HashList &root_catalog_list, | |
| 119 | const TraversalType type = Base::kBreadthFirst) { | ||
| 120 | // Push in reverse order for CatalogTraversal-like behavior | ||
| 121 | 2282 | HashList::const_reverse_iterator i = root_catalog_list.rbegin(); | |
| 122 | 2282 | const HashList::const_reverse_iterator iend = root_catalog_list.rend(); | |
| 123 | 2282 | bool has_pushed = false; | |
| 124 | { | ||
| 125 | 2282 | MutexLockGuard m(&catalogs_lock_); | |
| 126 |
3/4✓ Branch 2 taken 6919 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 4637 times.
✓ Branch 5 taken 2282 times.
|
6919 | for (; i != iend; ++i) { |
| 127 |
7/8✓ Branch 0 taken 4241 times.
✓ Branch 1 taken 396 times.
✓ Branch 4 taken 4241 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 1598 times.
✓ Branch 7 taken 2643 times.
✓ Branch 8 taken 1598 times.
✓ Branch 9 taken 3039 times.
|
4637 | if (this->no_repeat_history_ && catalogs_done_.Contains(*i)) { |
| 128 | 1598 | continue; | |
| 129 | } | ||
| 130 | |||
| 131 |
3/6✓ Branch 2 taken 3039 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 3039 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 3039 times.
✗ Branch 10 not taken.
|
3039 | CatalogJob *root_job = new CatalogJob("", *i, 0, 0); |
| 132 |
1/2✓ Branch 1 taken 3039 times.
✗ Branch 2 not taken.
|
3039 | PushJobUnlocked(root_job); |
| 133 | 3039 | has_pushed = true; | |
| 134 | } | ||
| 135 | 2282 | } | |
| 136 | // noop: no catalogs to traverse | ||
| 137 |
2/2✓ Branch 0 taken 611 times.
✓ Branch 1 taken 1671 times.
|
2282 | if (!has_pushed) { |
| 138 | 611 | return true; | |
| 139 | } | ||
| 140 | 1671 | effective_traversal_type_ = type; | |
| 141 | 1671 | effective_history_depth_ = Parameters::kNoHistory; | |
| 142 | 1671 | effective_timestamp_threshold_ = Parameters::kNoTimestampThreshold; | |
| 143 |
1/2✓ Branch 1 taken 1671 times.
✗ Branch 2 not taken.
|
1671 | bool result = DoTraverse(); |
| 144 | 1671 | effective_history_depth_ = this->default_history_depth_; | |
| 145 | 1671 | effective_timestamp_threshold_ = this->default_timestamp_threshold_; | |
| 146 | 1671 | 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 | 88 | bool TraverseRevision(const shash::Any &root_catalog_hash, | |
| 158 | const TraversalType type = Base::kBreadthFirst) { | ||
| 159 | 88 | effective_history_depth_ = Parameters::kNoHistory; | |
| 160 | 88 | effective_timestamp_threshold_ = Parameters::kNoTimestampThreshold; | |
| 161 | 88 | bool result = Traverse(root_catalog_hash, type); | |
| 162 | 88 | effective_history_depth_ = this->default_history_depth_; | |
| 163 | 88 | effective_timestamp_threshold_ = this->default_timestamp_threshold_; | |
| 164 | 88 | return result; | |
| 165 | } | ||
| 166 | |||
| 167 | protected: | ||
| 168 | 74286285 | 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 | 74286285 | *(reinterpret_cast<const uint32_t *>(key.digest) + 1)); | |
| 172 | } | ||
| 173 | |||
| 174 | 4154 | 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 | 4154 | threads_process_ = reinterpret_cast<pthread_t *>( | |
| 178 | 4154 | smalloc(sizeof(pthread_t) * num_threads_)); | |
| 179 |
2/2✓ Branch 0 taken 4770 times.
✓ Branch 1 taken 4154 times.
|
8924 | for (unsigned int i = 0; i < num_threads_; ++i) { |
| 180 | 4770 | int retval = pthread_create(&threads_process_[i], NULL, MainProcessQueue, | |
| 181 | this); | ||
| 182 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 4770 times.
|
4770 | if (retval != 0) |
| 183 | ✗ | PANIC(kLogStderr, "failed to create thread"); | |
| 184 | } | ||
| 185 | |||
| 186 |
2/2✓ Branch 0 taken 4770 times.
✓ Branch 1 taken 4154 times.
|
8924 | for (unsigned int i = 0; i < num_threads_; ++i) { |
| 187 | 4770 | int retval = pthread_join(threads_process_[i], NULL); | |
| 188 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 4770 times.
|
4770 | assert(retval == 0); |
| 189 | } | ||
| 190 | 4154 | free(threads_process_); | |
| 191 | |||
| 192 |
2/2✓ Branch 1 taken 44 times.
✓ Branch 2 taken 4110 times.
|
4154 | if (atomic_read32(&num_errors_)) |
| 193 | 44 | return false; | |
| 194 | |||
| 195 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 4110 times.
|
4110 | assert(catalogs_processing_.size() == 0); |
| 196 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 4110 times.
|
4110 | assert(pre_job_queue_.IsEmpty()); |
| 197 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 4110 times.
|
4110 | assert(post_job_queue_.IsEmpty()); |
| 198 | 4110 | return true; | |
| 199 | } | ||
| 200 | |||
| 201 | 4770 | static void *MainProcessQueue(void *data) { | |
| 202 | 4770 | CatalogTraversalParallel<ObjectFetcherT> *traversal = reinterpret_cast< | |
| 203 | CatalogTraversalParallel<ObjectFetcherT> *>(data); | ||
| 204 | CatalogJob *current_job; | ||
| 205 | while (true) { | ||
| 206 | 17704053 | current_job = traversal->post_job_queue_.TryPopFront(); | |
| 207 |
2/2✓ Branch 0 taken 1495245 times.
✓ Branch 1 taken 16211888 times.
|
17707133 | if (current_job != NULL) { |
| 208 | 1495245 | traversal->ProcessJobPost(current_job); | |
| 209 | } else { | ||
| 210 | 16211888 | current_job = traversal->pre_job_queue_.PopFront(); | |
| 211 | // NULL means the master thread tells us to finish | ||
| 212 |
2/2✓ Branch 1 taken 4770 times.
✓ Branch 2 taken 16207998 times.
|
16213032 | if (current_job->hash.IsNull()) { |
| 213 |
1/2✓ Branch 0 taken 4770 times.
✗ Branch 1 not taken.
|
4770 | delete current_job; |
| 214 | 4770 | break; | |
| 215 | } | ||
| 216 | 16207998 | traversal->ProcessJobPre(current_job); | |
| 217 | } | ||
| 218 | } | ||
| 219 | 4770 | return NULL; | |
| 220 | } | ||
| 221 | |||
| 222 | 4154 | void NotifyFinished() { | |
| 223 |
1/2✓ Branch 1 taken 4154 times.
✗ Branch 2 not taken.
|
4154 | shash::Any null_hash; |
| 224 | 4154 | null_hash.SetNull(); | |
| 225 |
2/2✓ Branch 0 taken 4770 times.
✓ Branch 1 taken 4154 times.
|
8924 | for (unsigned i = 0; i < num_threads_; ++i) { |
| 226 |
3/6✓ Branch 2 taken 4770 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 4770 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 4770 times.
✗ Branch 9 not taken.
|
4770 | CatalogJob *job = new CatalogJob("", null_hash, 0, 0); |
| 227 |
1/2✓ Branch 1 taken 4770 times.
✗ Branch 2 not taken.
|
4770 | pre_job_queue_.EnqueueFront(job); |
| 228 | } | ||
| 229 | 4154 | } | |
| 230 | |||
| 231 | 2483 | void PushJob(CatalogJob *job) { | |
| 232 | 2483 | MutexLockGuard m(&catalogs_lock_); | |
| 233 |
1/2✓ Branch 1 taken 2483 times.
✗ Branch 2 not taken.
|
2483 | PushJobUnlocked(job); |
| 234 | 2483 | } | |
| 235 | |||
| 236 | 16208394 | void PushJobUnlocked(CatalogJob *job) { | |
| 237 | 16208394 | catalogs_processing_.Insert(job->hash, job); | |
| 238 | 16208394 | pre_job_queue_.EnqueueFront(job); | |
| 239 | 16208394 | } | |
| 240 | |||
| 241 | 16207690 | void ProcessJobPre(CatalogJob *job) { | |
| 242 |
4/6✓ Branch 0 taken 16207690 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 16205314 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 44 times.
✓ Branch 6 taken 16205270 times.
|
16207690 | if (!this->PrepareCatalog(job)) { |
| 243 | 44 | atomic_inc32(&num_errors_); | |
| 244 |
1/2✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
|
44 | NotifyFinished(); |
| 245 | 14710157 | return; | |
| 246 | } | ||
| 247 |
2/2✓ Branch 0 taken 273 times.
✓ Branch 1 taken 16204997 times.
|
16205270 | if (job->ignore) { |
| 248 |
1/2✓ Branch 1 taken 273 times.
✗ Branch 2 not taken.
|
273 | FinalizeJob(job); |
| 249 | 273 | return; | |
| 250 | } | ||
| 251 |
1/2✓ Branch 1 taken 16159853 times.
✗ Branch 2 not taken.
|
16204997 | 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 | 16159853 | MutexLockGuard m(&catalogs_lock_); | |
| 257 |
2/2✓ Branch 0 taken 8124947 times.
✓ Branch 1 taken 8083086 times.
|
16208033 | if (effective_traversal_type_ == Base::kBreadthFirst) { |
| 258 |
1/2✓ Branch 1 taken 8124947 times.
✗ Branch 2 not taken.
|
8124947 | num_children = PushPreviousRevision(job) |
| 259 |
1/2✓ Branch 1 taken 8124947 times.
✗ Branch 2 not taken.
|
8124947 | + PushNestedCatalogs(job, catalog_list); |
| 260 | } else { | ||
| 261 |
1/2✓ Branch 1 taken 8083086 times.
✗ Branch 2 not taken.
|
8083086 | num_children = PushNestedCatalogs(job, catalog_list) |
| 262 |
1/2✓ Branch 1 taken 8083086 times.
✗ Branch 2 not taken.
|
8083086 | + PushPreviousRevision(job); |
| 263 | 8083086 | atomic_write32(&job->children_unprocessed, num_children); | |
| 264 | } | ||
| 265 |
3/6✓ Branch 0 taken 16208033 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 16208033 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 16208033 times.
|
16208033 | if (!this->CloseCatalog(false, job)) { |
| 266 | ✗ | atomic_inc32(&num_errors_); | |
| 267 | ✗ | NotifyFinished(); | |
| 268 | } | ||
| 269 | 16208033 | } | |
| 270 | |||
| 271 | // breadth-first: can post-process immediately | ||
| 272 | // depth-first: no children -> can post-process immediately | ||
| 273 |
4/4✓ Branch 0 taken 8082910 times.
✓ Branch 1 taken 8124903 times.
✓ Branch 2 taken 6587621 times.
✓ Branch 3 taken 1495289 times.
|
16207813 | if (effective_traversal_type_ == Base::kBreadthFirst || num_children == 0) { |
| 274 |
1/2✓ Branch 1 taken 14708740 times.
✗ Branch 2 not taken.
|
14712524 | ProcessJobPost(job); |
| 275 | 14708740 | return; | |
| 276 | } | ||
| 277 |
2/2✓ Branch 1 taken 1495289 times.
✓ Branch 2 taken 14709840 times.
|
16204029 | } |
| 278 | |||
| 279 | 16208033 | unsigned int PushNestedCatalogs(CatalogJob *job, | |
| 280 | const NestedCatalogList &catalog_list) { | ||
| 281 | 16208033 | typename NestedCatalogList::const_iterator i = catalog_list.begin(); | |
| 282 | 16208033 | typename NestedCatalogList::const_iterator iend = catalog_list.end(); | |
| 283 | 16208033 | unsigned int num_children = 0; | |
| 284 |
2/2✓ Branch 2 taken 16202742 times.
✓ Branch 3 taken 16208033 times.
|
32410775 | for (; i != iend; ++i) { |
| 285 |
7/8✓ Branch 0 taken 28694 times.
✓ Branch 1 taken 16174048 times.
✓ Branch 4 taken 28694 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2984 times.
✓ Branch 7 taken 25710 times.
✓ Branch 8 taken 2984 times.
✓ Branch 9 taken 16199758 times.
|
16202742 | if (this->no_repeat_history_ && catalogs_done_.Contains(i->hash)) { |
| 286 | 2984 | continue; | |
| 287 | } | ||
| 288 | |||
| 289 | CatalogJob *child; | ||
| 290 | 32399516 | if (!this->no_repeat_history_ | |
| 291 |
7/8✓ Branch 0 taken 25710 times.
✓ Branch 1 taken 16174048 times.
✓ Branch 4 taken 25710 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 25358 times.
✓ Branch 7 taken 352 times.
✓ Branch 8 taken 16199406 times.
✓ Branch 9 taken 352 times.
|
16199758 | || !catalogs_processing_.Lookup(i->hash, &child)) { |
| 292 |
2/2✓ Branch 0 taken 16132028 times.
✓ Branch 1 taken 67378 times.
|
16199406 | CatalogTN *parent = (this->no_close_) ? job->catalog : NULL; |
| 293 |
1/2✓ Branch 2 taken 16199406 times.
✗ Branch 3 not taken.
|
32398812 | child = new CatalogJob(i->mountpoint.ToString(), |
| 294 |
1/2✓ Branch 2 taken 16199406 times.
✗ Branch 3 not taken.
|
16199406 | i->hash, |
| 295 | 16199406 | job->tree_level + 1, | |
| 296 |
1/2✓ Branch 1 taken 16199406 times.
✗ Branch 2 not taken.
|
16199406 | job->history_depth, |
| 297 | parent); | ||
| 298 |
1/2✓ Branch 1 taken 16199406 times.
✗ Branch 2 not taken.
|
16199406 | PushJobUnlocked(child); |
| 299 | } | ||
| 300 | |||
| 301 |
2/2✓ Branch 0 taken 8081166 times.
✓ Branch 1 taken 8118592 times.
|
16199758 | if (effective_traversal_type_ == Base::kDepthFirst) { |
| 302 |
1/2✓ Branch 1 taken 8081166 times.
✗ Branch 2 not taken.
|
8081166 | child->RegisterListener(&CatalogTraversalParallel::OnChildFinished, |
| 303 | this, job); | ||
| 304 | } | ||
| 305 | 16199758 | ++num_children; | |
| 306 | } | ||
| 307 | 16208033 | 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 | 16208033 | unsigned int PushPreviousRevision(CatalogJob *job) { | |
| 315 | // only root catalogs are used for entering a previous revision (graph) | ||
| 316 |
2/2✓ Branch 1 taken 16199271 times.
✓ Branch 2 taken 8762 times.
|
16208033 | if (!job->catalog->IsRoot()) { |
| 317 | 16199271 | return 0; | |
| 318 | } | ||
| 319 | |||
| 320 |
1/2✓ Branch 1 taken 8762 times.
✗ Branch 2 not taken.
|
8762 | const shash::Any previous_revision = job->catalog->GetPreviousRevision(); |
| 321 |
2/2✓ Branch 1 taken 872 times.
✓ Branch 2 taken 7890 times.
|
8762 | if (previous_revision.IsNull()) { |
| 322 | 872 | return 0; | |
| 323 | } | ||
| 324 | |||
| 325 | // check if the next deeper history level is actually requested | ||
| 326 |
3/4✓ Branch 1 taken 7890 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 4424 times.
✓ Branch 4 taken 3466 times.
|
7890 | if (this->IsBelowPruningThresholds(*job, effective_history_depth_, |
| 327 | effective_timestamp_threshold_)) { | ||
| 328 | 4424 | return 0; | |
| 329 | } | ||
| 330 | |||
| 331 | 6932 | if (this->no_repeat_history_ | |
| 332 |
5/8✓ Branch 0 taken 2234 times.
✓ Branch 1 taken 1232 times.
✓ Branch 3 taken 2234 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 2234 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 3466 times.
|
3466 | && catalogs_done_.Contains(previous_revision)) { |
| 333 | ✗ | return 0; | |
| 334 | } | ||
| 335 | |||
| 336 | CatalogJob *prev_job; | ||
| 337 | 6932 | if (!this->no_repeat_history_ | |
| 338 |
5/8✓ Branch 0 taken 2234 times.
✓ Branch 1 taken 1232 times.
✓ Branch 3 taken 2234 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 2234 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 3466 times.
✗ Branch 8 not taken.
|
3466 | || !catalogs_processing_.Lookup(previous_revision, &prev_job)) { |
| 339 |
2/4✓ Branch 2 taken 3466 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 3466 times.
✗ Branch 6 not taken.
|
6932 | prev_job = new CatalogJob("", previous_revision, 0, |
| 340 |
1/2✓ Branch 1 taken 3466 times.
✗ Branch 2 not taken.
|
3466 | job->history_depth + 1); |
| 341 |
1/2✓ Branch 1 taken 3466 times.
✗ Branch 2 not taken.
|
3466 | PushJobUnlocked(prev_job); |
| 342 | } | ||
| 343 | |||
| 344 |
2/2✓ Branch 0 taken 660 times.
✓ Branch 1 taken 2806 times.
|
3466 | if (effective_traversal_type_ == Base::kDepthFirst) { |
| 345 |
1/2✓ Branch 1 taken 660 times.
✗ Branch 2 not taken.
|
660 | prev_job->RegisterListener(&CatalogTraversalParallel::OnChildFinished, |
| 346 | this, job); | ||
| 347 | } | ||
| 348 | 3466 | return 1; | |
| 349 | } | ||
| 350 | |||
| 351 | 16205393 | void ProcessJobPost(CatalogJob *job) { | |
| 352 | // Save time by keeping catalog open when suitable | ||
| 353 |
1/2✓ Branch 0 taken 16205393 times.
✗ Branch 1 not taken.
|
16205393 | if (job->catalog == NULL) { |
| 354 |
2/4✓ Branch 0 taken 16205393 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 16159237 times.
|
16205393 | if (!this->ReopenCatalog(job)) { |
| 355 | ✗ | atomic_inc32(&num_errors_); | |
| 356 | ✗ | NotifyFinished(); | |
| 357 | ✗ | return; | |
| 358 | } | ||
| 359 | } | ||
| 360 |
1/2✓ Branch 0 taken 16159237 times.
✗ Branch 1 not taken.
|
16159237 | if (serialize_callbacks_) { |
| 361 | 16159237 | MutexLockGuard m(&catalog_callback_lock_); | |
| 362 |
2/4✓ Branch 1 taken 16208033 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 16208033 times.
✗ Branch 5 not taken.
|
16208033 | this->NotifyListeners(job->GetCallbackData()); |
| 363 | 16208033 | } else { | |
| 364 | ✗ | this->NotifyListeners(job->GetCallbackData()); | |
| 365 | } | ||
| 366 |
2/2✓ Branch 0 taken 75785 times.
✓ Branch 1 taken 16132248 times.
|
16208033 | if (!this->no_close_) { |
| 367 |
2/4✓ Branch 0 taken 75785 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 75785 times.
|
75785 | if (!this->CloseCatalog(true, job)) { |
| 368 | ✗ | atomic_inc32(&num_errors_); | |
| 369 | ✗ | NotifyFinished(); | |
| 370 | ✗ | return; | |
| 371 | } | ||
| 372 | } | ||
| 373 | 16208033 | FinalizeJob(job); | |
| 374 | } | ||
| 375 | |||
| 376 | 16208262 | void FinalizeJob(CatalogJob *job) { | |
| 377 | { | ||
| 378 | 16208262 | MutexLockGuard m(&catalogs_lock_); | |
| 379 |
1/2✓ Branch 1 taken 16208306 times.
✗ Branch 2 not taken.
|
16208306 | catalogs_processing_.Erase(job->hash); |
| 380 |
1/2✓ Branch 1 taken 16208306 times.
✗ Branch 2 not taken.
|
16208306 | catalogs_done_.Insert(job->hash, true); |
| 381 | // No more catalogs to process -> finish | ||
| 382 |
1/2✓ Branch 2 taken 4110 times.
✗ Branch 3 not taken.
|
16212416 | if (catalogs_processing_.size() == 0 && pre_job_queue_.IsEmpty() |
| 383 |
5/6✓ Branch 0 taken 4110 times.
✓ Branch 1 taken 16204196 times.
✓ Branch 3 taken 4110 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 4110 times.
✓ Branch 6 taken 16204196 times.
|
16212416 | && post_job_queue_.IsEmpty()) { |
| 384 |
1/2✓ Branch 1 taken 4110 times.
✗ Branch 2 not taken.
|
4110 | NotifyFinished(); |
| 385 | } | ||
| 386 | 16208306 | } | |
| 387 |
2/2✓ Branch 0 taken 8083180 times.
✓ Branch 1 taken 8125038 times.
|
16208218 | if (effective_traversal_type_ == Base::kDepthFirst) { |
| 388 | 8083180 | job->WakeParents(); | |
| 389 | } | ||
| 390 |
2/2✓ Branch 0 taken 16207206 times.
✓ Branch 1 taken 264 times.
|
16207470 | delete job; |
| 391 | 16204874 | } | |
| 392 | |||
| 393 | 8081122 | void OnChildFinished(const int &a, CatalogJob *job) { | |
| 394 | // atomic_xadd32 returns value before subtraction -> needs to equal 1 | ||
| 395 |
2/2✓ Branch 1 taken 1495289 times.
✓ Branch 2 taken 6586273 times.
|
8081122 | if (atomic_xadd32(&job->children_unprocessed, -1) == 1) { |
| 396 | 1495289 | post_job_queue_.EnqueueFront(job); | |
| 397 | } | ||
| 398 | 8081562 | } | |
| 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 |