GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/ingestion/task.h
Date: 2025-12-28 02:35:52
Exec Total Coverage
Lines: 43 44 97.7%
Branches: 18 24 75.0%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #ifndef CVMFS_INGESTION_TASK_H_
6 #define CVMFS_INGESTION_TASK_H_
7
8 #include <errno.h>
9 #include <pthread.h>
10 #include <unistd.h>
11
12 #include <cassert>
13 #include <vector>
14
15 #include "util/exception.h"
16 #include "util/single_copy.h"
17 #include "util/tube.h"
18
19 /**
20 * Forward declaration of TubeConsumerGroup so that it can be used as a friend
21 * class to TubeConsumer.
22 */
23 template<typename ItemT>
24 class TubeConsumerGroup;
25
26
27 /**
28 * Base class for threads that processes items from a tube one by one. Concrete
29 * implementations overwrite the Process() method.
30 */
31 template<class ItemT>
32 class TubeConsumer : SingleCopy {
33 friend class TubeConsumerGroup<ItemT>;
34
35 public:
36 205952 virtual ~TubeConsumer() { }
37
38 protected:
39 204583 explicit TubeConsumer(Tube<ItemT> *tube) : tube_(tube) { }
40 virtual void Process(ItemT *item) = 0;
41 189358 virtual void OnTerminate() { }
42
43 Tube<ItemT> *tube_;
44
45 private:
46 193063 static void *MainConsumer(void *data) {
47 193063 TubeConsumer<ItemT> *consumer = reinterpret_cast<TubeConsumer<ItemT> *>(
48 data);
49
50 72387583 while (true) {
51 72580646 ItemT *item = consumer->tube_->PopFront();
52
2/2
✓ Branch 1 taken 96741 times.
✓ Branch 2 taken 39566291 times.
 71921587 if (item->IsQuitBeacon()) {
53
2/2
✓ Branch 0 taken 96726 times.
✓ Branch 1 taken 15 times.
 191840 delete item;
54 191502 break;
55 }
56 71621715 consumer->Process(item);
57 }
58 191502 consumer->OnTerminate();
59 191016 return NULL;
60 }
61 };
62
63
64 template<class ItemT>
65 class TubeConsumerGroup : SingleCopy {
66 public:
67 10499 TubeConsumerGroup() : is_active_(false) { }
68
69 10486 ~TubeConsumerGroup() {
70
2/2
✓ Branch 1 taken 102976 times.
✓ Branch 2 taken 5962 times.
 214796 for (unsigned i = 0; i < consumers_.size(); ++i)
71
1/2
✓ Branch 1 taken 102976 times.
✗ Branch 2 not taken.
 204310 delete consumers_[i];
72 10486 }
73
74 204583 void TakeConsumer(TubeConsumer<ItemT> *consumer) {
75
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 103113 times.
 204583 assert(!is_active_);
76 204583 consumers_.push_back(consumer);
77 204583 }
78
79 10139 void Spawn() {
80
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5789 times.
 10139 assert(!is_active_);
81 10139 unsigned N = consumers_.size();
82 10139 threads_.resize(N);
83
2/2
✓ Branch 0 taken 97353 times.
✓ Branch 1 taken 5789 times.
 203202 for (unsigned i = 0; i < N; ++i) {
84 193063 int retval = pthread_create(
85 193063 &threads_[i], NULL, TubeConsumer<ItemT>::MainConsumer, consumers_[i]);
86
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 97353 times.
 193063 if (retval != 0) {
87 ✗ PANIC(kLogStderr, "failed to create new thread (error: %d, pid: %d)",
88 errno, getpid());
89 }
90 }
91 10139 is_active_ = true;
92 10139 }
93
94 10126 void Terminate() {
95
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5782 times.
 10126 assert(is_active_);
96 10126 unsigned N = consumers_.size();
97
2/2
✓ Branch 0 taken 97216 times.
✓ Branch 1 taken 5782 times.
 202916 for (unsigned i = 0; i < N; ++i) {
98 192790 consumers_[i]->tube_->EnqueueBack(ItemT::CreateQuitBeacon());
99 }
100
2/2
✓ Branch 0 taken 97216 times.
✓ Branch 1 taken 5782 times.
 202916 for (unsigned i = 0; i < N; ++i) {
101 192790 int retval = pthread_join(threads_[i], NULL);
102
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 97216 times.
 192790 assert(retval == 0);
103 }
104 10126 is_active_ = false;
105 10126 }
106
107 1438 bool is_active() { return is_active_; }
108
109 private:
110 bool is_active_;
111 std::vector<TubeConsumer<ItemT> *> consumers_;
112 std::vector<pthread_t> threads_;
113 };
114
115 #endif // CVMFS_INGESTION_TASK_H_
116