leveldb 学习记录(四)Log文件

 1 enum RecordType {
 2   // Zero is reserved for preallocated files
 3   kZeroType = 0,
 4 
 5   kFullType = 1,
 6 
 7   // For fragments
 8   kFirstType = 2,
 9   kMiddleType = 3,
10   kLastType = 4
11 };
12 static const int kMaxRecordType = kLastType;
13 
14 static const int kBlockSize = 32768;
15 
16 // Header is checksum (4 bytes), type (1 byte), length (2 bytes).
17 static const int kHeaderSize = 4 + 1 + 2;
18 
19 }
20 }

 1 class Writer {
 2  public:
 3   // Create a writer that will append data to "*dest".
 4   // "*dest" must be initially empty.
 5   // "*dest" must remain live while this Writer is in use.
 6   explicit Writer(WritableFile* dest);
 7   ~Writer();
 8 
 9   Status AddRecord(const Slice& slice);
10 
11  private:
12   WritableFile* dest_;
13   int block_offset_;       // Current offset in block
14 
15   // crc32c values for all supported record types.  These are
16   // pre-computed to reduce the overhead of computing the crc of the
17   // record type stored in the header.
18   uint32_t type_crc_[kMaxRecordType + 1];
19 
20   Status EmitPhysicalRecord(RecordType type, const char* ptr, size_t length);
21 
22   // No copying allowed
23   Writer(const Writer&);
24   void operator=(const Writer&);
25 };

 1 Status Writer::AddRecord(const Slice& slice) {
 2   const char* ptr = slice.data();
 3   size_t left = slice.size();
 4 
 5   // Fragment the record if necessary and emit it.  Note that if slice
 6   // is empty, we still want to iterate once to emit a single
 7   // zero-length record
 8   Status s;
 9   bool begin = true;
10   do {
11     const int leftover = kBlockSize - block_offset_;        //剩余要填充的数据长度 是一个BLOCK的长度减去块内已经填充的长度
12     assert(leftover >= 0);
13     if (leftover < kHeaderSize) {                            //要填充的长度大于7 则在下一个BLOCK进行记录 (因为checksum 4字节  length2字节  type 1字节，光是记录信息已经需要7个字节)
14       // Switch to a new block
15       if (leftover > 0) {
16         // Fill the trailer (literal below relies on kHeaderSize being 7)
17         assert(kHeaderSize == 7);
18         dest_->Append(Slice("\x00\x00\x00\x00\x00\x00", leftover));
19       }
20       block_offset_ = 0;
21     }
22 
23     // Invariant: we never leave < kHeaderSize bytes in a block.
24     assert(kBlockSize - block_offset_ - kHeaderSize >= 0);
25 
26     const size_t avail = kBlockSize - block_offset_ - kHeaderSize;
27     const size_t fragment_length = (left < avail) ? left : avail;        //根据能否在本BLOCK填充完毕 选择填充长度为left 或者 avail
28 
29     RecordType type;
30     const bool end = (left == fragment_length);
31     if (begin && end) {                                //beg end在用一个BLOCK里 record的type肯定是FULL
32       type = kFullType;    
33     } else if (begin) {                                //本BLOCK只有beg 那么record的type    就是FIRST
34       type = kFirstType;
35     } else if (end) {                                //本BLOCK只有end 那么record的TYPE就是last
36       type = kLastType;
37     } else {
38       type = kMiddleType;                            //本BLOCK 没有beg end  那么record填充了整个BLOCK type是MIDDLE
39     }
40 
41     s = EmitPhysicalRecord(type, ptr, fragment_length);        //提交到log文件记录
42     ptr += fragment_length;
43     left -= fragment_length;
44     begin = false;
45   } while (s.ok() && left > 0);
46   return s;
47 }
48 
49 Status Writer::EmitPhysicalRecord(RecordType t, const char* ptr, size_t n) {
50   assert(n <= 0xffff);  // Must fit in two bytes
51   assert(block_offset_ + kHeaderSize + n <= kBlockSize);
52 
53   // Format the header
54   char buf[kHeaderSize];
55   buf[4] = static_cast<char>(n & 0xff);                //长度低8位
56   buf[5] = static_cast<char>(n >> 8);                //长度高8位
57   buf[6] = static_cast<char>(t);                    //type
58 
59   // Compute the crc of the record type and the payload.
60   uint32_t crc = crc32c::Extend(type_crc_[t], ptr, n);                //校验和
61   crc = crc32c::Mask(crc);                 // Adjust for storage
62   EncodeFixed32(buf, crc);
63 
64   // Write the header and the payload
65   Status s = dest_->Append(Slice(buf, kHeaderSize));        //数据信息写入
66   if (s.ok()) {
67     s = dest_->Append(Slice(ptr, n));                        //数据写入
68     if (s.ok()) {
69       s = dest_->Flush();
70     }
71   }
72   block_offset_ += kHeaderSize + n;
73   return s;
74 }

 1 class Reader {
 2  public:
 3   // Interface for reporting errors.
 4   class Reporter {
 5    public:
 6     virtual ~Reporter();
 7 
 8     // Some corruption was detected.  "size" is the approximate number
 9     // of bytes dropped due to the corruption.
10     virtual void Corruption(size_t bytes, const Status& status) = 0;
11   };
12 
13   // Create a reader that will return log records from "*file".
14   // "*file" must remain live while this Reader is in use.
15   //
16   // If "reporter" is non-NULL, it is notified whenever some data is
17   // dropped due to a detected corruption.  "*reporter" must remain
18   // live while this Reader is in use.
19   //
20   // If "checksum" is true, verify checksums if available.
21   //
22   // The Reader will start reading at the first record located at physical
23   // position >= initial_offset within the file.
24   Reader(SequentialFile* file, Reporter* reporter, bool checksum,
25          uint64_t initial_offset);
26 
27   ~Reader();
28 
29   // Read the next record into *record.  Returns true if read
30   // successfully, false if we hit end of the input.  May use
31   // "*scratch" as temporary storage.  The contents filled in *record
32   // will only be valid until the next mutating operation on this
33   // reader or the next mutation to *scratch.
34   bool ReadRecord(Slice* record, std::string* scratch);
35 
36   // Returns the physical offset of the last record returned by ReadRecord.
37   //
38   // Undefined before the first call to ReadRecord.
39   uint64_t LastRecordOffset();
40 
41  private:
42   SequentialFile* const file_;
43   Reporter* const reporter_;
44   bool const checksum_;
45   char* const backing_store_;
46   Slice buffer_;
47   bool eof_;   // Last Read() indicated EOF by returning < kBlockSize
48 
49   // Offset of the last record returned by ReadRecord.
50   uint64_t last_record_offset_;
51   // Offset of the first location past the end of buffer_.
52   uint64_t end_of_buffer_offset_;
53 
54   // Offset at which to start looking for the first record to return
55   uint64_t const initial_offset_;
56 
57   // Extend record types with the following special values
58   enum {
59     kEof = kMaxRecordType + 1,
60     // Returned whenever we find an invalid physical record.
61     // Currently there are three situations in which this happens:
62     // * The record has an invalid CRC (ReadPhysicalRecord reports a drop)
63     // * The record is a 0-length record (No drop is reported)
64     // * The record is below constructor‘s initial_offset (No drop is reported)
65     kBadRecord = kMaxRecordType + 2
66   };
67 
68   // Skips all blocks that are completely before "initial_offset_".
69   //
70   // Returns true on success. Handles reporting.
71   bool SkipToInitialBlock();
72 
73   // Return type, or one of the preceding special values
74   unsigned int ReadPhysicalRecord(Slice* result);
75 
76   // Reports dropped bytes to the reporter.
77   // buffer_ must be updated to remove the dropped bytes prior to invocation.
78   void ReportCorruption(size_t bytes, const char* reason);
79   void ReportDrop(size_t bytes, const Status& reason);
80 
81   // No copying allowed
82   Reader(const Reader&);
83   void operator=(const Reader&);
84 };

bool Reader::ReadRecord(Slice* record, std::string* scratch) {
  if (last_record_offset_ < initial_offset_) {                    //实际上整个工程中initial_offset_一直为0  ,
    if (!SkipToInitialBlock()) {                                //block_start_location圆整为包含initial_offset_的BLOCK的偏移
      return false;
    }
  }

  scratch->clear();
  record->clear();
  bool in_fragmented_record = false;
  // Record offset of the logical record that we‘re reading
  // 0 is a dummy value to make compilers happy
  uint64_t prospective_record_offset = 0;

  Slice fragment;
  while (true) {
    uint64_t physical_record_offset = end_of_buffer_offset_ - buffer_.size();
    const unsigned int record_type = ReadPhysicalRecord(&fragment);
    switch (record_type) {
      case kFullType:                                //一次性读取FULL类型的record 直接返回成功
        if (in_fragmented_record) {
          // Handle bug in earlier versions of log::Writer where
          // it could emit an empty kFirstType record at the tail end
          // of a block followed by a kFullType or kFirstType record
          // at the beginning of the next block.
          if (scratch->empty()) {
            in_fragmented_record = false;
          } else {
            ReportCorruption(scratch->size(), "partial record without end(1)");
          }
        }
        prospective_record_offset = physical_record_offset;
        scratch->clear();
        *record = fragment;
        last_record_offset_ = prospective_record_offset;
        return true;

      case kFirstType:                        //读取到FIRST类型的record  string.assign  然后继续
        if (in_fragmented_record) {
          // Handle bug in earlier versions of log::Writer where
          // it could emit an empty kFirstType record at the tail end
          // of a block followed by a kFullType or kFirstType record
          // at the beginning of the next block.
          if (scratch->empty()) {
            in_fragmented_record = false;
          } else {
            ReportCorruption(scratch->size(), "partial record without end(2)");
          }
        }
        prospective_record_offset = physical_record_offset;
        scratch->assign(fragment.data(), fragment.size());
        in_fragmented_record = true;
        break;

      case kMiddleType:                            //读取到MIDDLE类型的record  string.append  然后继续
        if (!in_fragmented_record) {
          ReportCorruption(fragment.size(),
                           "missing start of fragmented record(1)");
        } else {
          scratch->append(fragment.data(), fragment.size());
        }
        break;

      case kLastType:                            //读取到LAST 类型record string.append
        if (!in_fragmented_record) {
          ReportCorruption(fragment.size(),
                           "missing start of fragmented record(2)");
        } else {
          scratch->append(fragment.data(), fragment.size());
          *record = Slice(*scratch);
          last_record_offset_ = prospective_record_offset;
          return true;
        }
        break;

      case kEof:
        if (in_fragmented_record) {
          ReportCorruption(scratch->size(), "partial record without end(3)");
          scratch->clear();
        }
        return false;

      case kBadRecord:
        if (in_fragmented_record) {
          ReportCorruption(scratch->size(), "error in middle of record");
          in_fragmented_record = false;
          scratch->clear();
        }
        break;

      default: {
        char buf[40];
        snprintf(buf, sizeof(buf), "unknown record type %u", record_type);
        ReportCorruption(
            (fragment.size() + (in_fragmented_record ? scratch->size() : 0)),
            buf);
        in_fragmented_record = false;
        scratch->clear();
        break;
      }
    }
  }
  return false;
}