ceph/src/rocksdb/db/blob/blob_garbage_meter.cc

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5
   6 #include "db/blob/blob_garbage_meter.h"
   7
   8 #include "db/blob/blob_index.h"
   9 #include "db/blob/blob_log_format.h"
  10 #include "db/dbformat.h"
  11
  12 namespace ROCKSDB_NAMESPACE {
  13
  14 Status BlobGarbageMeter::ProcessInFlow(const Slice& key, const Slice& value) {
  15   uint64_t blob_file_number = kInvalidBlobFileNumber;
  16   uint64_t bytes = 0;
  17
  18   const Status s = Parse(key, value, &blob_file_number, &bytes);
  19   if (!s.ok()) {
  20     return s;
  21   }
  22
  23   if (blob_file_number == kInvalidBlobFileNumber) {
  24     return Status::OK();
  25   }
  26
  27   flows_[blob_file_number].AddInFlow(bytes);
  28
  29   return Status::OK();
  30 }
  31
  32 Status BlobGarbageMeter::ProcessOutFlow(const Slice& key, const Slice& value) {
  33   uint64_t blob_file_number = kInvalidBlobFileNumber;
  34   uint64_t bytes = 0;
  35
  36   const Status s = Parse(key, value, &blob_file_number, &bytes);
  37   if (!s.ok()) {
  38     return s;
  39   }
  40
  41   if (blob_file_number == kInvalidBlobFileNumber) {
  42     return Status::OK();
  43   }
  44
  45   // Note: in order to measure the amount of additional garbage, we only need to
  46   // track the outflow for preexisting files, i.e. those that also had inflow.
  47   // (Newly written files would only have outflow.)
  48   auto it = flows_.find(blob_file_number);
  49   if (it == flows_.end()) {
  50     return Status::OK();
  51   }
  52
  53   it->second.AddOutFlow(bytes);
  54
  55   return Status::OK();
  56 }
  57
  58 Status BlobGarbageMeter::Parse(const Slice& key, const Slice& value,
  59                                uint64_t* blob_file_number, uint64_t* bytes) {
  60   assert(blob_file_number);
  61   assert(*blob_file_number == kInvalidBlobFileNumber);
  62   assert(bytes);
  63   assert(*bytes == 0);
  64
  65   ParsedInternalKey ikey;
  66
  67   {
  68     constexpr bool log_err_key = false;
  69     const Status s = ParseInternalKey(key, &ikey, log_err_key);
  70     if (!s.ok()) {
  71       return s;
  72     }
  73   }
  74
  75   if (ikey.type != kTypeBlobIndex) {
  76     return Status::OK();
  77   }
  78
  79   BlobIndex blob_index;
  80
  81   {
  82     const Status s = blob_index.DecodeFrom(value);
  83     if (!s.ok()) {
  84       return s;
  85     }
  86   }
  87
  88   if (blob_index.IsInlined() || blob_index.HasTTL()) {
  89     return Status::Corruption("Unexpected TTL/inlined blob index");
  90   }
  91
  92   *blob_file_number = blob_index.file_number();
  93   *bytes =
  94       blob_index.size() +
  95       BlobLogRecord::CalculateAdjustmentForRecordHeader(ikey.user_key.size());
  96
  97   return Status::OK();
  98 }
  99
 100 }  // namespace ROCKSDB_NAMESPACE