GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/receiver/payload_processor.cc
Date: 2026-08-23 02:40:52
Exec Total Coverage
Lines: 34 137 24.8%
Branches: 20 150 13.3%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #include "payload_processor.h"
6
7 #include <fcntl.h>
8 #include <unistd.h>
9
10 #include <vector>
11
12 #include "params.h"
13 #include "util/logging.h"
14 #include "util/posix.h"
15 #include "util/string.h"
16
17 namespace {
18
19 const size_t kConsumerBuffer = 10 * 1024 * 1024; // 10 MB
20
21 }
22
23 namespace receiver {
24
25 ✗ FileInfo::FileInfo()
26 ✗ : handle(NULL)
27 ✗ , total_size(0)
28 ✗ , current_size(0)
29 ✗ , hash_context()
30 ✗ , hash_buffer() { }
31
32 ✗ FileInfo::FileInfo(const ObjectPackBuild::Event &event)
33 ✗ : handle(NULL)
34 ✗ , total_size(event.size)
35 ✗ , current_size(0)
36 ✗ , hash_context(shash::ContextPtr(event.id.algorithm))
37 ✗ , hash_buffer(hash_context.size, 0) {
38 ✗ hash_context.buffer = &hash_buffer[0];
39 ✗ shash::Init(hash_context);
40 }
41
42 ✗ FileInfo::FileInfo(const FileInfo &other)
43 ✗ : handle(other.handle)
44 ✗ , total_size(other.total_size)
45 ✗ , current_size(other.current_size)
46 ✗ , hash_context(other.hash_context)
47 ✗ , hash_buffer(other.hash_buffer) {
48 ✗ hash_context.buffer = &hash_buffer[0];
49 }
50
51 ✗ FileInfo &FileInfo::operator=(const FileInfo &other) {
52 ✗ handle = other.handle;
53 ✗ total_size = other.total_size;
54 ✗ current_size = other.current_size;
55 ✗ hash_context = other.hash_context;
56 ✗ hash_buffer = other.hash_buffer;
57 ✗ hash_context.buffer = &hash_buffer[0];
58
59 ✗ return *this;
60 }
61
62 69 PayloadProcessor::PayloadProcessor()
63 69 : pending_files_()
64 69 , current_repo_()
65 69 , uploader_()
66 69 , temp_dir_()
67 69 , num_errors_(0)
68 138 , statistics_(nullptr) { }
69
70 138 PayloadProcessor::~PayloadProcessor() { }
71
72 69 PayloadProcessor::Result PayloadProcessor::Process(
73 int fdin, const std::string &header_digest, const std::string &path,
74 uint64_t header_size) {
75
1/2
✓ Branch 3 taken 69 times.
✗ Branch 4 not taken.
 69 LogCvmfs(kLogReceiver, kLogSyslog,
76 "PayloadProcessor - lease_path: %s, header digest: %s, header "
77 "size: %ld",
78 path.c_str(), header_digest.c_str(), header_size);
79
80 69 const size_t first_slash_idx = path.find('/', 0);
81
82
1/2
✓ Branch 1 taken 69 times.
✗ Branch 2 not taken.
 69 current_repo_ = path.substr(0, first_slash_idx);
83
84
1/2
✓ Branch 1 taken 69 times.
✗ Branch 2 not taken.
 69 const Result init_result = Initialize();
85
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 69 times.
 69 if (init_result != kSuccess) {
86 ✗ return init_result;
87 }
88
89 // Set up object pack deserialization
90
1/2
✓ Branch 2 taken 69 times.
✗ Branch 3 not taken.
 69 const shash::Any digest = shash::MkFromHexPtr(shash::HexPtr(header_digest));
91
92
1/2
✓ Branch 1 taken 69 times.
✗ Branch 2 not taken.
 69 ObjectPackConsumer deserializer(digest, header_size);
93
1/2
✓ Branch 1 taken 69 times.
✗ Branch 2 not taken.
 69 deserializer.RegisterListener(&PayloadProcessor::ConsumerEventCallback, this);
94
95 69 int nb = 0;
96 69 ObjectPackBuild::State consumer_state = ObjectPackBuild::kStateContinue;
97
1/2
✓ Branch 2 taken 69 times.
✗ Branch 3 not taken.
 69 std::vector<unsigned char> buffer(kConsumerBuffer, 0);
98 do {
99
1/2
✓ Branch 3 taken 69 times.
✗ Branch 4 not taken.
 69 nb = read(fdin, &buffer[0], buffer.size());
100
1/2
✓ Branch 2 taken 69 times.
✗ Branch 3 not taken.
 69 consumer_state = deserializer.ConsumeNext(nb, &buffer[0]);
101
1/2
✓ Branch 0 taken 69 times.
✗ Branch 1 not taken.
 69 if (consumer_state != ObjectPackBuild::kStateContinue
102
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 69 times.
 69 && consumer_state != ObjectPackBuild::kStateDone) {
103 ✗ LogCvmfs(kLogReceiver, kLogSyslogErr,
104 "PayloadProcessor - error: %d encountered when consuming object "
105 "pack.",
106 consumer_state);
107 ✗ break;
108 }
109
2/4
✓ Branch 0 taken 69 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 69 times.
 69 } while (nb > 0 && consumer_state != ObjectPackBuild::kStateDone);
110
111
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 69 times.
 69 assert(pending_files_.empty());
112
113
1/2
✓ Branch 1 taken 69 times.
✗ Branch 2 not taken.
 69 const Result res = Finalize();
114
115
1/2
✓ Branch 1 taken 69 times.
✗ Branch 2 not taken.
 69 deserializer.UnregisterListeners();
116
117 69 return res;
118 69 }
119
120 ✗ void PayloadProcessor::ConsumerEventCallback(
121 const ObjectPackBuild::Event &event) {
122 ✗ std::string path("");
123
124 ✗ if (event.object_type == ObjectPack::kCas) {
125 ✗ path = event.id.MakePath();
126 ✗ } else if (event.object_type == ObjectPack::kNamed) {
127 ✗ path = event.object_name;
128 } else {
129 // kEmpty - this is an error.
130 ✗ LogCvmfs(kLogReceiver, kLogSyslogErr,
131 "PayloadProcessor - error: Event received with unknown object.");
132 ✗ num_errors_++;
133 ✗ return;
134 }
135
136 ✗ const FileIterator it = pending_files_.find(event.id);
137 ✗ if (it == pending_files_.end()) {
138 // Schedule file upload if it's not being uploaded yet.
139 // Uploaders later check if the file is already present
140 // in the upstream storage and will not upload it twice.
141 ✗ FileInfo info(event);
142 // info.handle is later deleted by FinalizeStreamedUpload
143 ✗ info.handle = uploader_->InitStreamedUpload(NULL);
144 ✗ pending_files_[event.id] = info;
145 }
146
147 ✗ FileInfo &info = pending_files_[event.id];
148
149 ✗ void *buf_copied = smalloc(event.buf_size);
150 ✗ memcpy(buf_copied, event.buf, event.buf_size);
151 ✗ const upload::AbstractUploader::UploadBuffer buf(event.buf_size, buf_copied);
152 ✗ uploader_->ScheduleUpload(
153 info.handle, buf,
154 ✗ upload::AbstractUploader::MakeClosure(
155 &PayloadProcessor::OnUploadJobComplete, this, buf_copied));
156
157 ✗ shash::Update(static_cast<const unsigned char *>(event.buf),
158 ✗ event.buf_size,
159 info.hash_context);
160
161 ✗ info.current_size += event.buf_size;
162
163 ✗ if (info.current_size == info.total_size) {
164 ✗ shash::Any file_hash(event.id.algorithm);
165 ✗ shash::Final(info.hash_context, &file_hash);
166
167 ✗ if (file_hash != event.id) {
168 ✗ LogCvmfs(
169 kLogReceiver, kLogSyslogErr,
170 "PayloadProcessor - error: Hash mismatch for unpacked file: event "
171 "size: %ld, file size: %ld, event hash: %s, file hash: %s",
172 ✗ event.size, info.current_size, event.id.ToString(true).c_str(),
173 ✗ file_hash.ToString(true).c_str());
174 ✗ num_errors_++;
175 ✗ return;
176 }
177 // override final remote path if not CAS object
178 ✗ if (event.object_type == ObjectPack::kNamed) {
179 ✗ info.handle->remote_path = path;
180 }
181 ✗ uploader_->ScheduleCommit(info.handle, event.id);
182
183 ✗ pending_files_.erase(event.id);
184 }
185 }
186
187 ✗ void PayloadProcessor::OnUploadJobComplete(
188 const upload::UploaderResults &results, void *buffer) {
189 ✗ free(buffer);
190 }
191
192 20 void PayloadProcessor::SetStatistics(perf::Statistics *st) {
193
1/2
✓ Branch 2 taken 20 times.
✗ Branch 3 not taken.
 60 statistics_ = std::unique_ptr<perf::StatisticsTemplate>(
194
2/4
✓ Branch 1 taken 20 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 20 times.
✗ Branch 5 not taken.
 40 new perf::StatisticsTemplate("publish", st));
195 20 }
196
197 ✗ PayloadProcessor::Result PayloadProcessor::Initialize() {
198 ✗ Params params;
199 ✗ if (!GetParamsFromFile(current_repo_, &params)) {
200 ✗ LogCvmfs(
201 kLogReceiver, kLogSyslogErr,
202 "PayloadProcessor - error: Could not get configuration parameters.");
203 ✗ return kOtherError;
204 }
205
206 const std::string spooler_temp_dir = GetSpoolerTempDir(
207 ✗ params.spooler_configuration);
208 ✗ assert(!spooler_temp_dir.empty());
209 ✗ assert(MkdirDeep(spooler_temp_dir + "/receiver", 0770, true));
210 ✗ temp_dir_ = std::unique_ptr<RaiiTempDir>(
211 ✗ RaiiTempDir::Create(spooler_temp_dir + "/receiver/payload_processor"));
212
213 const upload::SpoolerDefinition definition(
214 params.spooler_configuration, params.hash_alg, params.compression_alg,
215 ✗ params.generate_legacy_bulk_chunks, params.use_file_chunking,
216 params.min_chunk_size, params.avg_chunk_size, params.max_chunk_size,
217 ✗ "dummy_token", "dummy_key");
218
219 ✗ uploader_.reset();
220
221 // configure the uploader environment
222 ✗ uploader_ = std::unique_ptr<upload::AbstractUploader>(
223 ✗ upload::AbstractUploader::Construct(definition));
224 ✗ if (uploader_.get() == nullptr) {
225 ✗ LogCvmfs(kLogSpooler, kLogWarning,
226 "Failed to initialize backend upload "
227 "facility in PayloadProcessor.");
228 ✗ return kUploaderError;
229 }
230
231 ✗ if (statistics_.get() != nullptr) {
232 ✗ uploader_->InitCounters(statistics_.get());
233 }
234
235 ✗ return kSuccess;
236 }
237
238 ✗ PayloadProcessor::Result PayloadProcessor::Finalize() {
239 ✗ uploader_->WaitForUpload();
240 ✗ temp_dir_.reset();
241
242 ✗ const unsigned num_uploader_errors = uploader_->GetNumberOfErrors();
243 ✗ uploader_->TearDown();
244 ✗ if (num_uploader_errors > 0) {
245 ✗ LogCvmfs(kLogReceiver, kLogSyslogErr,
246 "PayloadProcessor - error: Uploader - %d upload(s) failed.",
247 num_uploader_errors);
248 ✗ return kUploaderError;
249 }
250
251 ✗ if (GetNumErrors() > 0) {
252 ✗ LogCvmfs(kLogReceiver, kLogSyslogErr,
253 "PayloadProcessor - error: %d unpacking error(s).",
254 GetNumErrors());
255 ✗ return kOtherError;
256 }
257
258 ✗ return kSuccess;
259 }
260
261 } // namespace receiver
262