Home | History | Annotate | Download | only in payload_generator 
      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 <brillo/secure_blob.h>
     32 
     33 #include "update_engine/common/utils.h"
     34 #include "update_engine/payload_generator/delta_diff_utils.h"
     35 
     36 using std::vector;
     37 
     38 namespace chromeos_update_engine {
     39 
     40 namespace {
     41 
     42 const size_t kDefaultFullChunkSize = 1024 * 1024;  // 1 MiB
     43 
     44 // This class encapsulates a full update chunk processing thread work. The
     45 // processor reads a chunk of data from the input file descriptor and compresses
     46 // it. The processor will destroy itself when the work is done.
     47 class ChunkProcessor : public base::DelegateSimpleThread::Delegate {
     48  public:
     49   // Read a chunk of |size| bytes from |fd| starting at offset |offset|.
     50   ChunkProcessor(const PayloadVersion& version,
     51                  int fd,
     52                  off_t offset,
     53                  size_t size,
     54                  BlobFileWriter* blob_file,
     55                  AnnotatedOperation* aop)
     56       : version_(version),
     57         fd_(fd),
     58         offset_(offset),
     59         size_(size),
     60         blob_file_(blob_file),
     61         aop_(aop) {}
     62   // We use a default move constructor since all the data members are POD types.
     63   ChunkProcessor(ChunkProcessor&&) = default;
     64   ~ChunkProcessor() override = default;
     65 
     66   // Overrides DelegateSimpleThread::Delegate.
     67   // Run() handles the read from |fd| in a thread-safe way, and stores the
     68   // new operation to generate the region starting at |offset| of size |size|
     69   // in the output operation |aop|. The associated blob data is stored in
     70   // |blob_fd| and |blob_file_size| is updated.
     71   void Run() override;
     72 
     73  private:
     74   bool ProcessChunk();
     75 
     76   // Work parameters.
     77   const PayloadVersion& version_;
     78   int fd_;
     79   off_t offset_;
     80   size_t size_;
     81   BlobFileWriter* blob_file_;
     82   AnnotatedOperation* aop_;
     83 
     84   DISALLOW_COPY_AND_ASSIGN(ChunkProcessor);
     85 };
     86 
     87 void ChunkProcessor::Run() {
     88   if (!ProcessChunk()) {
     89     LOG(ERROR) << "Error processing region at " << offset_ << " of size "
     90                << size_;
     91   }
     92 }
     93 
     94 bool ChunkProcessor::ProcessChunk() {
     95   brillo::Blob buffer_in_(size_);
     96   brillo::Blob op_blob;
     97   ssize_t bytes_read = -1;
     98   TEST_AND_RETURN_FALSE(utils::PReadAll(fd_,
     99                                         buffer_in_.data(),
    100                                         buffer_in_.size(),
    101                                         offset_,
    102                                         &bytes_read));
    103   TEST_AND_RETURN_FALSE(bytes_read == static_cast<ssize_t>(size_));
    104 
    105   InstallOperation_Type op_type;
    106   TEST_AND_RETURN_FALSE(diff_utils::GenerateBestFullOperation(
    107       buffer_in_, version_, &op_blob, &op_type));
    108 
    109   aop_->op.set_type(op_type);
    110   TEST_AND_RETURN_FALSE(aop_->SetOperationBlob(op_blob, blob_file_));
    111   return true;
    112 }
    113 
    114 }  // namespace
    115 
    116 bool FullUpdateGenerator::GenerateOperations(
    117     const PayloadGenerationConfig& config,
    118     const PartitionConfig& old_part,
    119     const PartitionConfig& new_part,
    120     BlobFileWriter* blob_file,
    121     vector<AnnotatedOperation>* aops) {
    122   TEST_AND_RETURN_FALSE(new_part.ValidateExists());
    123 
    124   // FullUpdateGenerator requires a positive chunk_size, otherwise there will
    125   // be only one operation with the whole partition which should not be allowed.
    126   // For performance reasons, we force a small default hard limit of 1 MiB. This
    127   // limit can be changed in the config, and we will use the smaller of the two
    128   // soft/hard limits.
    129   size_t full_chunk_size;
    130   if (config.hard_chunk_size >= 0) {
    131     full_chunk_size = std::min(static_cast<size_t>(config.hard_chunk_size),
    132                                config.soft_chunk_size);
    133   } else {
    134     full_chunk_size = std::min(kDefaultFullChunkSize, config.soft_chunk_size);
    135     LOG(INFO) << "No chunk_size provided, using the default chunk_size for the "
    136               << "full operations: " << full_chunk_size << " bytes.";
    137   }
    138   TEST_AND_RETURN_FALSE(full_chunk_size > 0);
    139   TEST_AND_RETURN_FALSE(full_chunk_size % config.block_size == 0);
    140 
    141   size_t chunk_blocks = full_chunk_size / config.block_size;
    142   size_t max_threads = std::max(sysconf(_SC_NPROCESSORS_ONLN), 4L);
    143   LOG(INFO) << "Compressing partition " << new_part.name
    144             << " from " << new_part.path << " splitting in chunks of "
    145             << chunk_blocks << " blocks (" << config.block_size
    146             << " bytes each) using " << max_threads << " threads";
    147 
    148   int in_fd = open(new_part.path.c_str(), O_RDONLY, 0);
    149   TEST_AND_RETURN_FALSE(in_fd >= 0);
    150   ScopedFdCloser in_fd_closer(&in_fd);
    151 
    152   // We potentially have all the ChunkProcessors in memory but only
    153   // |max_threads| will actually hold a block in memory while we process.
    154   size_t partition_blocks = new_part.size / config.block_size;
    155   size_t num_chunks = (partition_blocks + chunk_blocks - 1) / chunk_blocks;
    156   aops->resize(num_chunks);
    157   vector<ChunkProcessor> chunk_processors;
    158   chunk_processors.reserve(num_chunks);
    159   blob_file->SetTotalBlobs(num_chunks);
    160 
    161   for (size_t i = 0; i < num_chunks; ++i) {
    162     size_t start_block = i * chunk_blocks;
    163     // The last chunk could be smaller.
    164     size_t num_blocks = std::min(chunk_blocks,
    165                                  partition_blocks - i * chunk_blocks);
    166 
    167     // Preset all the static information about the operations. The
    168     // ChunkProcessor will set the rest.
    169     AnnotatedOperation* aop = aops->data() + i;
    170     aop->name = base::StringPrintf("<%s-operation-%" PRIuS ">",
    171                                    new_part.name.c_str(), i);
    172     Extent* dst_extent = aop->op.add_dst_extents();
    173     dst_extent->set_start_block(start_block);
    174     dst_extent->set_num_blocks(num_blocks);
    175 
    176     chunk_processors.emplace_back(
    177         config.version,
    178         in_fd,
    179         static_cast<off_t>(start_block) * config.block_size,
    180         num_blocks * config.block_size,
    181         blob_file,
    182         aop);
    183   }
    184 
    185   // Thread pool used for worker threads.
    186   base::DelegateSimpleThreadPool thread_pool("full-update-generator",
    187                                              max_threads);
    188   thread_pool.Start();
    189   for (ChunkProcessor& processor : chunk_processors)
    190     thread_pool.AddWork(&processor);
    191   thread_pool.JoinAll();
    192 
    193   // All the work done, disable logging.
    194   blob_file->SetTotalBlobs(0);
    195 
    196   // All the operations must have a type set at this point. Otherwise, a
    197   // ChunkProcessor failed to complete.
    198   for (const AnnotatedOperation& aop : *aops) {
    199     if (!aop.op.has_type())
    200       return false;
    201   }
    202   return true;
    203 }
    204 
    205 }  // namespace chromeos_update_engine
    206