GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/ingestion/item.h
Date: 2026-10-25 02:40:14
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 <atomic>
12 #include <cassert>
13 #include <memory>
14 #include <string>
15 #include <vector>
16
17 #include "compression/compression.h"
18 #include "crypto/hash.h"
19 #include "file_chunk.h"
20 #include "ingestion/chunk_detector.h"
21 #include "ingestion/ingestion_source.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 116712 static FileItem *CreateQuitBeacon() {
48
1/2
✓ Branch 2 taken 116712 times.
✗ Branch 3 not taken.
 116712 const std::string quit_marker = std::string(1, kQuitBeaconMarker);
49 std::unique_ptr<FileIngestionSource> source(
50
2/4
✓ Branch 1 taken 116712 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 116712 times.
✗ Branch 5 not taken.
 116712 new FileIngestionSource(quit_marker));
51
2/4
✓ Branch 2 taken 116712 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 116712 times.
✗ Branch 6 not taken.
 233424 return new FileItem(source.release());
52 116712 }
53 16617480 bool IsQuitBeacon() {
54
10/21
✓ Branch 1 taken 16610283 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 116645 times.
✓ Branch 5 taken 16493647 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 116709 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 116771 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 116066 times.
✓ Branch 13 taken 705 times.
✓ Branch 14 taken 116709 times.
✓ Branch 15 taken 16493709 times.
✓ Branch 17 taken 16610860 times.
✗ Branch 18 not taken.
✗ Branch 20 not taken.
✗ Branch 21 not taken.
✗ Branch 23 not taken.
✗ Branch 24 not taken.
 16617480 return (path().length() == 1) && (path()[0] == kQuitBeaconMarker);
55 }
56
57 41482625 std::string path() { return source_->GetPath(); }
58 16737407 uint64_t size() { return size_; }
59 26963941 Xor32Detector *chunk_detector() { return &chunk_detector_; }
60 41247826 shash::Any bulk_hash() { return bulk_hash_; }
61 16510792 zlib::Algorithms compression_algorithm() { return compression_algorithm_; }
62 16751323 shash::Algorithms hash_algorithm() { return hash_algorithm_; }
63 24741420 shash::Suffix hash_suffix() { return hash_suffix_; }
64 16486930 bool may_have_chunks() { return may_have_chunks_; }
65 8251811 bool has_legacy_bulk_chunk() { return has_legacy_bulk_chunk_; }
66
67 8246471 void set_size(uint64_t val) { size_ = val; }
68 8245698 void set_may_have_chunks(bool val) { may_have_chunks_ = val; }
69 8242565 void set_is_fully_chunked() { is_fully_chunked_.fetch_add(1); }
70 16535758 bool is_fully_chunked() { return is_fully_chunked_.load() != 0; }
71 16501903 uint64_t nchunks_in_fly() { return nchunks_in_fly_.load(); }
72
73 16500892 uint64_t GetNumChunks() { return chunks_.size(); }
74 8252543 FileChunkList *GetChunksPtr() { return &chunks_; }
75
76 8249597 bool Open() { return source_->Open(); }
77 18269284 ssize_t Read(void *buffer, size_t nbyte) {
78 18269284 return source_->Read(buffer, nbyte);
79 }
80 8249630 bool Close() { return source_->Close(); }
81 8248673 bool GetSize(uint64_t *size) { return source_->GetSize(size); }
82
83 // Called by ChunkItem constructor, decremented when a chunk is registered
84 8497823 void IncNchunksInFly() { nchunks_in_fly_.fetch_add(1); }
85 void RegisterChunk(const FileChunk &file_chunk);
86 8267591 bool IsProcessed() {
87
4/4
✓ Branch 1 taken 8264711 times.
✓ Branch 2 taken 2880 times.
✓ Branch 4 taken 8252639 times.
✓ Branch 5 taken 12072 times.
 16532302 return is_fully_chunked() && (nchunks_in_fly_.load() == 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 std::atomic<int64_t> nchunks_in_fly_;
110 /**
111 * Switches to true once all of the file has been through the chunking
112 * stage
113 */
114 std::atomic<int32_t> 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 9227093 bool IsSolePiece() {
131
6/6
✓ Branch 0 taken 975081 times.
✓ Branch 1 taken 8252012 times.
✓ Branch 2 taken 747 times.
✓ Branch 3 taken 974334 times.
✓ Branch 5 taken 144 times.
✓ Branch 6 taken 603 times.
 9227093 return !is_bulk_chunk_ && (offset_ == 0) && (size_ == file_item_->size());
132 }
133
134 960912 bool is_bulk_chunk() { return is_bulk_chunk_; }
135 8267639 FileItem *file_item() { return file_item_; }
136 9493097 uint64_t offset() { return offset_; }
137 8507735 uint64_t size() { return size_; }
138 22575703 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 20999500 shash::ContextPtr hash_ctx() { return hash_ctx_; }
146 24795285 shash::Any *hash_ptr() { return &hash_value_; }
147
148 8492219 void set_size(uint64_t val) {
149
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8492219 times.
 8492219 assert(size_ == 0);
150 8492219 size_ = val;
151 8492219 }
152 8265149 void set_upload_handle(upload::UploadStreamHandle *val) {
153
2/4
✓ Branch 0 taken 8265182 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 8265182 times.
✗ Branch 3 not taken.
 8265149 assert((upload_handle_ == NULL) && (val != NULL));
154 8265182 upload_handle_ = val;
155 8265182 }
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 102848 times.
✗ Branch 3 not taken.
 102848 static BlockItem *CreateQuitBeacon() { return new BlockItem(NULL); }
196 81582433 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 42929783 bool IsFull() { return size_ == capacity_; }
211
212 25237806924 unsigned char *data() { return data_; }
213 25187457448 uint32_t capacity() { return capacity_; }
214 136647940 uint32_t size() { return size_; }
215 21544228 void set_size(uint32_t val) {
216
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 21544228 times.
 21544228 assert(val <= capacity_);
217 21544228 size_ = val;
218 21544228 }
219
220 87317995 BlockType type() { return type_; }
221 118129989 int64_t tag() { return tag_; }
222 39716156 FileItem *file_item() { return file_item_; }
223 114225205 ChunkItem *chunk_item() { return chunk_item_; }
224 16614049 static uint64_t managed_bytes() { return managed_bytes_.load(); }
225
226 private:
227 /**
228 * Total capacity of all BlockItem()
229 */
230 static std::atomic<int64_t> 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