GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/ingestion/item.h
Date: 2026-08-30 02:40:36
Exec Total Coverage
Lines: 63 63 100.0%
Branches: 30 51 58.8%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #ifndef CVMFS_INGESTION_ITEM_H_
6 #define CVMFS_INGESTION_ITEM_H_
7
8 #include <pthread.h>
9 #include <stdint.h>
10
11 #include <cassert>
12 #include <memory>
13 #include <string>
14 #include <vector>
15
16 #include "compression/compression.h"
17 #include "crypto/hash.h"
18 #include "file_chunk.h"
19 #include "ingestion/chunk_detector.h"
20 #include "ingestion/ingestion_source.h"
21 #include "util/atomic.h"
22 #include "util/single_copy.h"
23
24 namespace upload {
25 struct UploadStreamHandle;
26 }
27
28 class ItemAllocator;
29
30 /**
31 * Carries the information necessary to compress and checksum a file. During
32 * processing, the bulk chunk and the chunks_ vector are filled.
33 */
34 class FileItem : SingleCopy {
35 public:
36 explicit FileItem(IngestionSource *source,
37 uint64_t min_chunk_size = 4 * 1024 * 1024,
38 uint64_t avg_chunk_size = 8 * 1024 * 1024,
39 uint64_t max_chunk_size = 16 * 1024 * 1024,
40 zlib::Algorithms compression_algorithm = zlib::kZlibDefault,
41 shash::Algorithms hash_algorithm = shash::kSha1,
42 shash::Suffix hash_suffix = shash::kSuffixNone,
43 bool may_have_chunks = true,
44 bool has_legacy_bulk_chunk = false);
45 ~FileItem();
46
47 144108 static FileItem *CreateQuitBeacon() {
48
1/2
✓ Branch 2 taken 144108 times.
✗ Branch 3 not taken.
 144108 const std::string quit_marker = std::string(1, kQuitBeaconMarker);
49 std::unique_ptr<FileIngestionSource> source(
50
2/4
✓ Branch 1 taken 144108 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 144108 times.
✗ Branch 5 not taken.
 144108 new FileIngestionSource(quit_marker));
51
2/4
✓ Branch 2 taken 144108 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 144108 times.
✗ Branch 6 not taken.
 288216 return new FileItem(source.release());
52 144108 }
53 22145951 bool IsQuitBeacon() {
54
10/21
✓ Branch 1 taken 22137998 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 144848 times.
✓ Branch 5 taken 21993595 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 143649 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 143697 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 142866 times.
✓ Branch 13 taken 831 times.
✓ Branch 14 taken 143746 times.
✓ Branch 15 taken 21993546 times.
✓ Branch 17 taken 22138276 times.
✗ Branch 18 not taken.
✗ Branch 20 not taken.
✗ Branch 21 not taken.
✗ Branch 23 not taken.
✗ Branch 24 not taken.
 22145951 return (path().length() == 1) && (path()[0] == kQuitBeaconMarker);
55 }
56
57 55286673 std::string path() { return source_->GetPath(); }
58 22024595 uint64_t size() { return size_; }
59 35346806 Xor32Detector *chunk_detector() { return &chunk_detector_; }
60 54988441 shash::Any bulk_hash() { return bulk_hash_; }
61 22013278 zlib::Algorithms compression_algorithm() { return compression_algorithm_; }
62 22040978 shash::Algorithms hash_algorithm() { return hash_algorithm_; }
63 32991015 shash::Suffix hash_suffix() { return hash_suffix_; }
64 21981148 bool may_have_chunks() { return may_have_chunks_; }
65 11002873 bool has_legacy_bulk_chunk() { return has_legacy_bulk_chunk_; }
66
67 10992135 void set_size(uint64_t val) { size_ = val; }
68 10989048 void set_may_have_chunks(bool val) { may_have_chunks_ = val; }
69 10989653 void set_is_fully_chunked() { atomic_inc32(&is_fully_chunked_); }
70 11021783 bool is_fully_chunked() { return atomic_read32(&is_fully_chunked_) != 0; }
71 11000943 uint64_t nchunks_in_fly() { return atomic_read64(&nchunks_in_fly_); }
72
73 22002190 uint64_t GetNumChunks() { return chunks_.size(); }
74 11003553 FileChunkList *GetChunksPtr() { return &chunks_; }
75
76 10998353 bool Open() { return source_->Open(); }
77 24344952 ssize_t Read(void *buffer, size_t nbyte) {
78 24344952 return source_->Read(buffer, nbyte);
79 }
80 10997561 bool Close() { return source_->Close(); }
81 10995317 bool GetSize(uint64_t *size) { return source_->GetSize(size); }
82
83 // Called by ChunkItem constructor, decremented when a chunk is registered
84 11040417 void IncNchunksInFly() { atomic_inc64(&nchunks_in_fly_); }
85 void RegisterChunk(const FileChunk &file_chunk);
86 11021747 bool IsProcessed() {
87
4/4
✓ Branch 1 taken 11018449 times.
✓ Branch 2 taken 3298 times.
✓ Branch 4 taken 11003609 times.
✓ Branch 5 taken 14840 times.
 11021747 return is_fully_chunked() && (atomic_read64(&nchunks_in_fly_) == 0);
88 }
89
90 private:
91 static const uint64_t kSizeUnknown = uint64_t(-1);
92 static const char kQuitBeaconMarker = '\0';
93
94 std::unique_ptr<IngestionSource> source_;
95 const zlib::Algorithms compression_algorithm_;
96 const shash::Algorithms hash_algorithm_;
97 const shash::Suffix hash_suffix_;
98 const bool has_legacy_bulk_chunk_;
99
100 uint64_t size_;
101 bool may_have_chunks_;
102
103 Xor32Detector chunk_detector_;
104 shash::Any bulk_hash_;
105 FileChunkList chunks_;
106 /**
107 * Number of chunks created but not yet uploaded and registered
108 */
109 atomic_int64 nchunks_in_fly_;
110 /**
111 * Switches to true once all of the file has been through the chunking
112 * stage
113 */
114 atomic_int32 is_fully_chunked_;
115 pthread_mutex_t lock_;
116 };
117
118
119 /**
120 * A chunk stores the state of compression and hashing contexts as the blocks
121 * move through the pipeline. A chunk can be a "bulk chunk" corresponding to
122 * the processed data of an entire file, or it can be a partial chunk of a
123 * (large) input file.
124 */
125 class ChunkItem : SingleCopy {
126 public:
127 ChunkItem(FileItem *file_item, uint64_t offset);
128
129 void MakeBulkChunk();
130 11141055 bool IsSolePiece() {
131
6/6
✓ Branch 0 taken 138194 times.
✓ Branch 1 taken 11002861 times.
✓ Branch 2 taken 532 times.
✓ Branch 3 taken 137662 times.
✓ Branch 5 taken 18 times.
✓ Branch 6 taken 514 times.
 11141055 return !is_bulk_chunk_ && (offset_ == 0) && (size_ == file_item_->size());
132 }
133
134 120114 bool is_bulk_chunk() { return is_bulk_chunk_; }
135 11021753 FileItem *file_item() { return file_item_; }
136 11207385 uint64_t offset() { return offset_; }
137 11051765 uint64_t size() { return size_; }
138 28200446 upload::UploadStreamHandle *upload_handle() { return upload_handle_; }
139 // An active zlib compression stream requires 256kB of memory. Therefore,
140 // we create it only for the absolutely necessary duration and free the space
141 // afterwards.
142 zlib::Compressor *GetCompressor();
143 void ReleaseCompressor();
144
145 27999352 shash::ContextPtr hash_ctx() { return hash_ctx_; }
146 33053239 shash::Any *hash_ptr() { return &hash_value_; }
147
148 11034211 void set_size(uint64_t val) {
149
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 11034211 times.
 11034211 assert(size_ == 0);
150 11034211 size_ = val;
151 11034211 }
152 11019107 void set_upload_handle(upload::UploadStreamHandle *val) {
153
2/4
✓ Branch 0 taken 11019151 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 11019151 times.
✗ Branch 3 not taken.
 11019107 assert((upload_handle_ == NULL) && (val != NULL));
154 11019151 upload_handle_ = val;
155 11019151 }
156
157 private:
158 FileItem *file_item_;
159 uint64_t offset_;
160 /**
161 * The size of a chunk is not defined before the corresponding stop block
162 * has been dispatched.
163 */
164 uint64_t size_;
165 bool is_bulk_chunk_;
166 /**
167 * Deleted by the uploader.
168 */
169 upload::UploadStreamHandle *upload_handle_;
170 std::unique_ptr<zlib::Compressor> compressor_;
171 shash::ContextPtr hash_ctx_;
172 shash::Any hash_value_;
173 unsigned char hash_ctx_buffer_[shash::kMaxContextSize];
174 };
175
176
177 /**
178 * A block is an item of work in the pipeline. A sequence of data blocks
179 * followed by a stop block constitutes a Chunk. A sequence of Chunks in turn
180 * build constitute a file.
181 * A block that carries data must have a non-zero-length payload.
182 */
183 class BlockItem : SingleCopy {
184 public:
185 enum BlockType {
186 kBlockHollow,
187 kBlockData,
188 kBlockStop,
189 };
190
191 explicit BlockItem(ItemAllocator *allocator);
192 BlockItem(int64_t tag, ItemAllocator *allocator);
193 ~BlockItem();
194
195
1/2
✓ Branch 2 taken 127984 times.
✗ Branch 3 not taken.
 127984 static BlockItem *CreateQuitBeacon() { return new BlockItem(NULL); }
196 106227899 bool IsQuitBeacon() { return type_ == kBlockHollow; }
197
198 void MakeStop();
199 void MakeData(uint32_t capacity);
200 void MakeDataMove(BlockItem *other);
201 void MakeDataCopy(const unsigned char *data, uint32_t size);
202 void SetFileItem(FileItem *item);
203 void SetChunkItem(ChunkItem *item);
204 // Free data and reset to hollow block
205 void Reset();
206
207 uint32_t Write(void *buf, uint32_t count);
208
209 bool IsEmpty() { return size_ == 0; }
210 57172904 bool IsFull() { return size_ == capacity_; }
211
212 24733749924 unsigned char *data() { return data_; }
213 24670292516 uint32_t capacity() { return capacity_; }
214 170467113 uint32_t size() { return size_; }
215 28667667 void set_size(uint32_t val) {
216
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 28667667 times.
 28667667 assert(val <= capacity_);
217 28667667 size_ = val;
218 28667667 }
219
220 111025274 BlockType type() { return type_; }
221 155737460 int64_t tag() { return tag_; }
222 52357100 FileItem *file_item() { return file_item_; }
223 146589221 ChunkItem *chunk_item() { return chunk_item_; }
224 11071341 static uint64_t managed_bytes() { return atomic_read64(&managed_bytes_); }
225
226 private:
227 /**
228 * Total capacity of all BlockItem()
229 */
230 static atomic_int64 managed_bytes_;
231
232 // Forget pointer to the data
233 void Discharge();
234
235 ItemAllocator *allocator_;
236 BlockType type_;
237
238 /**
239 * Blocks with the same tag need to be processed sequentially. That is, no
240 * two threads of the same pipeline stage must operate on blocks of the same
241 * tag. The tags roughly correspond to chunks.
242 * Tags can (and should) be set exactly once in the life time of a block.
243 */
244 int64_t tag_;
245
246 /**
247 * Can be set exactly once.
248 */
249 FileItem *file_item_;
250 ChunkItem *chunk_item_;
251
252 /**
253 * Managed by ItemAllocator
254 */
255 unsigned char *data_;
256 uint32_t capacity_;
257 uint32_t size_;
258 };
259
260 #endif // CVMFS_INGESTION_ITEM_H_
261