full_update_generator.cc revision 625406cee9a90ac2ed895f480286b7f0e8497f38 
1//
2// Copyright (C) 2012 The Android Open Source Project
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8//      http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15//
16
17#include "update_engine/payload_generator/full_update_generator.h"
18
19#include <fcntl.h>
20#include <inttypes.h>
21
22#include <algorithm>
23#include <deque>
24#include <memory>
25
26#include <base/format_macros.h>
27#include <base/strings/string_util.h>
28#include <base/strings/stringprintf.h>
29#include <base/synchronization/lock.h>
30#include <base/threading/simple_thread.h>
31#include <chromeos/secure_blob.h>
32
33#include "update_engine/bzip.h"
34#include "update_engine/utils.h"
35
36using std::string;
37using std::vector;
38
39namespace chromeos_update_engine {
40
41namespace {
42
43const size_t kDefaultFullChunkSize = 1024 * 1024;  // 1 MiB
44
45// This class encapsulates a full update chunk processing thread work. The
46// processor reads a chunk of data from the input file descriptor and compresses
47// it. The processor will destroy itself when the work is done.
48class ChunkProcessor : public base::DelegateSimpleThread::Delegate {
49 public:
50  // Read a chunk of |size| bytes from |fd| starting at offset |offset|.
51  ChunkProcessor(int fd, off_t offset, size_t size,
52                 BlobFileWriter* blob_file, AnnotatedOperation* aop)
53    : fd_(fd),
54      offset_(offset),
55      size_(size),
56      blob_file_(blob_file),
57      aop_(aop) {}
58  // We use a default move constructor since all the data members are POD types.
59  ChunkProcessor(ChunkProcessor&&) = default;
60  ~ChunkProcessor() override = default;
61
62  // Overrides DelegateSimpleThread::Delegate.
63  // Run() handles the read from |fd| in a thread-safe way, and stores the
64  // new operation to generate the region starting at |offset| of size |size|
65  // in the output operation |aop|. The associated blob data is stored in
66  // |blob_fd| and |blob_file_size| is updated.
67  void Run() override;
68
69 private:
70  bool ProcessChunk();
71
72  // Stores the operation blob in the |blob_fd_| and updates the
73  // |blob_file_size| accordingly.
74  // This method is thread-safe since it uses a mutex to access the file.
75  // Returns the data offset where the data was written to.
76  off_t StoreBlob(const chromeos::Blob& blob);
77
78  // Work parameters.
79  int fd_;
80  off_t offset_;
81  size_t size_;
82  BlobFileWriter* blob_file_;
83  AnnotatedOperation* aop_;
84
85  DISALLOW_COPY_AND_ASSIGN(ChunkProcessor);
86};
87
88void ChunkProcessor::Run() {
89  if (!ProcessChunk()) {
90    LOG(ERROR) << "Error processing region at " << offset_ << " of size "
91               << size_;
92  }
93}
94
95bool ChunkProcessor::ProcessChunk() {
96  chromeos::Blob buffer_in_(size_);
97  chromeos::Blob op_blob;
98  ssize_t bytes_read = -1;
99  TEST_AND_RETURN_FALSE(utils::PReadAll(fd_,
100                                        buffer_in_.data(),
101                                        buffer_in_.size(),
102                                        offset_,
103                                        &bytes_read));
104  TEST_AND_RETURN_FALSE(bytes_read == static_cast<ssize_t>(size_));
105  TEST_AND_RETURN_FALSE(BzipCompress(buffer_in_, &op_blob));
106
107  InstallOperation_Type op_type = InstallOperation::REPLACE_BZ;
108
109  if (op_blob.size() >= buffer_in_.size()) {
110    // A REPLACE is cheaper than a REPLACE_BZ in this case.
111    op_blob = std::move(buffer_in_);
112    op_type = InstallOperation::REPLACE;
113  }
114
115  TEST_AND_RETURN_FALSE(aop_->SetOperationBlob(&op_blob, blob_file_));
116  aop_->op.set_type(op_type);
117  return true;
118}
119
120}  // namespace
121
122bool FullUpdateGenerator::GenerateOperations(
123    const PayloadGenerationConfig& config,
124    const PartitionConfig& old_part,
125    const PartitionConfig& new_part,
126    BlobFileWriter* blob_file,
127    vector<AnnotatedOperation>* aops) {
128  TEST_AND_RETURN_FALSE(new_part.ValidateExists());
129
130  // FullUpdateGenerator requires a positive chunk_size, otherwise there will
131  // be only one operation with the whole partition which should not be allowed.
132  // For performance reasons, we force a small default hard limit of 1 MiB. This
133  // limit can be changed in the config, and we will use the smaller of the two
134  // soft/hard limits.
135  size_t full_chunk_size;
136  if (config.hard_chunk_size >= 0) {
137    full_chunk_size = std::min(static_cast<size_t>(config.hard_chunk_size),
138                               config.soft_chunk_size);
139  } else {
140    full_chunk_size = std::min(kDefaultFullChunkSize, config.soft_chunk_size);
141    LOG(INFO) << "No chunk_size provided, using the default chunk_size for the "
142              << "full operations: " << full_chunk_size << " bytes.";
143  }
144  TEST_AND_RETURN_FALSE(full_chunk_size > 0);
145  TEST_AND_RETURN_FALSE(full_chunk_size % config.block_size == 0);
146
147  size_t chunk_blocks = full_chunk_size / config.block_size;
148  size_t max_threads = std::max(sysconf(_SC_NPROCESSORS_ONLN), 4L);
149  LOG(INFO) << "Compressing partition " << new_part.name
150            << " from " << new_part.path << " splitting in chunks of "
151            << chunk_blocks << " blocks (" << config.block_size
152            << " bytes each) using " << max_threads << " threads";
153
154  int in_fd = open(new_part.path.c_str(), O_RDONLY, 0);
155  TEST_AND_RETURN_FALSE(in_fd >= 0);
156  ScopedFdCloser in_fd_closer(&in_fd);
157
158  // We potentially have all the ChunkProcessors in memory but only
159  // |max_threads| will actually hold a block in memory while we process.
160  size_t partition_blocks = new_part.size / config.block_size;
161  size_t num_chunks = (partition_blocks + chunk_blocks - 1) / chunk_blocks;
162  aops->resize(num_chunks);
163  vector<ChunkProcessor> chunk_processors;
164  chunk_processors.reserve(num_chunks);
165  blob_file->SetTotalBlobs(num_chunks);
166
167  for (size_t i = 0; i < num_chunks; ++i) {
168    size_t start_block = i * chunk_blocks;
169    // The last chunk could be smaller.
170    size_t num_blocks = std::min(chunk_blocks,
171                                 partition_blocks - i * chunk_blocks);
172
173    // Preset all the static information about the operations. The
174    // ChunkProcessor will set the rest.
175    AnnotatedOperation* aop = aops->data() + i;
176    aop->name = base::StringPrintf("<%s-operation-%" PRIuS ">",
177                                   new_part.name.c_str(), i);
178    Extent* dst_extent = aop->op.add_dst_extents();
179    dst_extent->set_start_block(start_block);
180    dst_extent->set_num_blocks(num_blocks);
181
182    chunk_processors.emplace_back(
183        in_fd,
184        static_cast<off_t>(start_block) * config.block_size,
185        num_blocks * config.block_size,
186        blob_file,
187        aop);
188  }
189
190  // Thread pool used for worker threads.
191  base::DelegateSimpleThreadPool thread_pool("full-update-generator",
192                                             max_threads);
193  thread_pool.Start();
194  for (ChunkProcessor& processor : chunk_processors)
195    thread_pool.AddWork(&processor);
196  thread_pool.JoinAll();
197  // All the operations must have a type set at this point. Otherwise, a
198  // ChunkProcessor failed to complete.
199  for (const AnnotatedOperation& aop : *aops) {
200    if (!aop.op.has_type())
201      return false;
202  }
203  return true;
204}
205
206}  // namespace chromeos_update_engine
207