GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/ingestion/item.h
Date: 2026-09-13 02:40:16
Exec Total Coverage
Lines: 63 63 100.0%
Branches: 31 51 60.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 59016 static FileItem *CreateQuitBeacon() {
48
1/2
✓ Branch 2 taken 59016 times.
✗ Branch 3 not taken.
 59016 const std::string quit_marker = std::string(1, kQuitBeaconMarker);
49 std::unique_ptr<FileIngestionSource> source(
50
2/4
✓ Branch 1 taken 59016 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 59016 times.
✗ Branch 5 not taken.
 59016 new FileIngestionSource(quit_marker));
51
2/4
✓ Branch 2 taken 59016 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 59016 times.
✗ Branch 6 not taken.
 118032 return new FileItem(source.release());
52 59016 }
53 10060337 bool IsQuitBeacon() {
54
11/21
✓ Branch 1 taken 10056529 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 59532 times.
✓ Branch 5 taken 9997062 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 58885 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 58882 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 58502 times.
✓ Branch 13 taken 380 times.
✓ Branch 14 taken 58879 times.
✓ Branch 15 taken 9997065 times.
✓ Branch 17 taken 10055878 times.
✓ Branch 18 taken 163 times.
✗ Branch 20 not taken.
✗ Branch 21 not taken.
✗ Branch 23 not taken.
✗ Branch 24 not taken.
 10060337 return (path().length() == 1) && (path()[0] == kQuitBeaconMarker);
55 }
56
57 25117207 std::string path() { return source_->GetPath(); }
58 10075777 uint64_t size() { return size_; }
59 16210819 Xor32Detector *chunk_detector() { return &chunk_detector_; }
60 24995669 shash::Any bulk_hash() { return bulk_hash_; }
61 10005098 zlib::Algorithms compression_algorithm() { return compression_algorithm_; }
62 10085050 shash::Algorithms hash_algorithm() { return hash_algorithm_; }
63 14994543 shash::Suffix hash_suffix() { return hash_suffix_; }
64 9989266 bool may_have_chunks() { return may_have_chunks_; }
65 5001077 bool has_legacy_bulk_chunk() { return has_legacy_bulk_chunk_; }
66
67 4996749 void set_size(uint64_t val) { size_ = val; }
68 4994828 void set_may_have_chunks(bool val) { may_have_chunks_ = val; }
69 4994521 void set_is_fully_chunked() { atomic_inc32(&is_fully_chunked_); }
70 5010477 bool is_fully_chunked() { return atomic_read32(&is_fully_chunked_) != 0; }
71 5000733 uint64_t nchunks_in_fly() { return atomic_read64(&nchunks_in_fly_); }
72
73 10001238 uint64_t GetNumChunks() { return chunks_.size(); }
74 5001673 FileChunkList *GetChunksPtr() { return &chunks_; }
75
76 4999409 bool Open() { return source_->Open(); }
77 11071783 ssize_t Read(void *buffer, size_t nbyte) {
78 11071783 return source_->Read(buffer, nbyte);
79 }
80 4998369 bool Close() { return source_->Close(); }
81 4997929 bool GetSize(uint64_t *size) { return source_->GetSize(size); }
82
83 // Called by ChunkItem constructor, decremented when a chunk is registered
84 5083937 void IncNchunksInFly() { atomic_inc64(&nchunks_in_fly_); }
85 void RegisterChunk(const FileChunk &file_chunk);
86 5010381 bool IsProcessed() {
87
4/4
✓ Branch 1 taken 5008889 times.
✓ Branch 2 taken 1492 times.
✓ Branch 4 taken 5001725 times.
✓ Branch 5 taken 7164 times.
 5010381 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 5330073 bool IsSolePiece() {
131
6/6
✓ Branch 0 taken 328708 times.
✓ Branch 1 taken 5001365 times.
✓ Branch 2 taken 348 times.
✓ Branch 3 taken 328360 times.
✓ Branch 5 taken 48 times.
✓ Branch 6 taken 300 times.
 5330073 return !is_bulk_chunk_ && (offset_ == 0) && (size_ == file_item_->size());
132 }
133
134 320304 bool is_bulk_chunk() { return is_bulk_chunk_; }
135 5010397 FileItem *file_item() { return file_item_; }
136 5423909 uint64_t offset() { return offset_; }
137 5090429 uint64_t size() { return size_; }
138 13253486 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 12726014 shash::ContextPtr hash_ctx() { return hash_ctx_; }
146 15025547 shash::Any *hash_ptr() { return &hash_value_; }
147
148 5081045 void set_size(uint64_t val) {
149
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5081045 times.
 5081045 assert(size_ == 0);
150 5081045 size_ = val;
151 5081045 }
152 5008881 void set_upload_handle(upload::UploadStreamHandle *val) {
153
2/4
✓ Branch 0 taken 5008881 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 5008881 times.
✗ Branch 3 not taken.
 5008881 assert((upload_handle_ == NULL) && (val != NULL));
154 5008881 upload_handle_ = val;
155 5008881 }
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 52160 times.
✗ Branch 3 not taken.
 52160 static BlockItem *CreateQuitBeacon() { return new BlockItem(NULL); }
196 48854806 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 26000646 bool IsFull() { return size_ == capacity_; }
211
212 8955653716 unsigned char *data() { return data_; }
213 8925959161 uint32_t capacity() { return capacity_; }
214 79244317 uint32_t size() { return size_; }
215 13030642 void set_size(uint32_t val) {
216
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 13030642 times.
 13030642 assert(val <= capacity_);
217 13030642 size_ = val;
218 13030642 }
219
220 50871094 BlockType type() { return type_; }
221 71221258 int64_t tag() { return tag_; }
222 23938029 FileItem *file_item() { return file_item_; }
223 67936773 ChunkItem *chunk_item() { return chunk_item_; }
224 5033637 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